FastAPI 到关系数据库

connection 生成器将 FastAPI 连接到 Python 关系数据库项目，通过 FastAPI 依赖项将类型化的 SQLModel 会话注入到您的路由处理程序中。

前置条件

在使用此生成器之前，请确保您已具备：

1. 一个 FastAPI 项目（使用 py#api --framework=fastapi 生成），请参阅 py#api 指南
2. 一个 py#rdb 项目

使用方法

运行生成器

CLI

pnpm

pnpm nx g @aws/nx-plugin:connection

yarn

yarn nx g @aws/nx-plugin:connection

npm

npx nx g @aws/nx-plugin:connection

bun

bunx nx g @aws/nx-plugin:connection

您还可以执行试运行以查看哪些文件会被更改

pnpm

pnpm nx g @aws/nx-plugin:connection --dry-run

yarn

yarn nx g @aws/nx-plugin:connection --dry-run

npm

npx nx g @aws/nx-plugin:connection --dry-run

bun

bunx nx g @aws/nx-plugin:connection --dry-run

VSCode

1. 安装 Nx Console VSCode Plugin 如果您尚未安装
2. 在VSCode中打开Nx控制台
3. 点击 Generate (UI) 在"Common Nx Commands"部分
4. 搜索 @aws/nx-plugin - connection
5. 填写必需参数
6. 点击 Generate

选择您的 FastAPI 项目作为源，选择您的关系数据库项目作为目标。

选项

参数	类型	默认值	描述
sourceProject 必需	string	-	源项目
targetProject 必需	string	-	要连接到的目标项目
sourceComponent	string	-	要从其连接的源组件（组件名称、相对于源项目根目录的路径或生成器 ID）。使用 '.' 显式选择项目作为源。
targetComponent	string	-	要连接到的目标组件（组件名称、相对于目标项目根目录的路径或生成器 ID）。使用 '.' 显式选择项目作为目标。
preferInstallDependencies	boolean	true	是否在生成器运行后优先安装依赖项。设置为 false 可在批量运行多个生成器时延迟安装（如果后续生成器需要计算 Nx 项目图，仍会运行安装）；在最后统一安装一次。

生成器输出

生成器会修改您的 FastAPI 项目：

• 文件夹packages/my_api

  • project.json 添加从 dev 到数据库的 dev 目标的依赖
  • pyproject.toml 将数据库包添加为工作区依赖项
  • 文件夹my_api

    • 文件夹dependencies

      • my_db.py 用于数据库会话的 FastAPI MyDbSession 依赖项

在路由处理程序中使用数据库

此生成器配置了一个可注入的 FastAPI Dependency，您可以在路由处理程序中使用它：

packages/my_api/my_api/api.py

from sqlmodel import select
from my_api.dependencies.my_db import MyDbSession
from my_scope.my_db.models.example import ExampleModel

@app.get("/examples")
async def list_examples(my_db: MyDbSession):
    return (await my_db.execute(select(ExampleModel))).all()

@app.post("/examples")
async def create_example(name: str, my_db: MyDbSession):
    item = ExampleModel(name=name)
    my_db.add(item)
    await my_db.commit()
    await my_db.refresh(item)
    return item

FastAPI 会自动为每个请求打开一个新会话，并在处理程序返回时关闭它。

基础设施

为了允许 FastAPI Lambda 函数在运行时连接到数据库，它必须部署到与数据库相同的 VPC 中，并被授予网络和 IAM 访问权限。

CDK

在您的应用程序堆栈中，将 API 部署到与数据库相同的 VPC 中，然后调用 allowDefaultPortFrom 和 grantConnect 来打开网络路径并授予 Lambda 处理程序 IAM rds-db:connect 权限：

packages/infra/src/stacks/application-stack.ts

import { MyDatabase } from '@my-scope/common-constructs';

const db = new MyDatabase(this, 'Db', { vpc, ... });

const api = new MyApi(this, 'Api', {
  integrations: MyApi.defaultIntegrations(this)
    .withDefaultOptions({
      vpc,
      vpcSubnets: { subnetType: SubnetType.PRIVATE_WITH_EGRESS },
    })
    .build(),
});

Object.entries(api.integrations).forEach(([operation, integration]) => {
  db.allowDefaultPortFrom(integration.handler, `Allow ${operation} to connect to the database`);
  db.grantConnect(integration.handler);
});

将 API Lambda 函数部署到具有出口的私有子网，而不是私有隔离子网。在运行时，session_context() 从 AWS AppConfig 检索数据库配置，这是一个需要出站互联网访问的公共 AWS 服务端点。

注意

此示例授予 API 中的每个处理程序访问权限，但如果只有某些处理程序需要访问，最好单独配置。

Terraform

将 API 部署到与数据库相同的 VPC 中，通过 additional_iam_policy_statements 授予它 rds-db:connect 权限，并使用一对安全组规则打开网络路径：

下面的示例引用了 Python 关系数据库部署指南中的 VPC 资源：aws_subnet.database 子网没有互联网路由，而 aws_subnet.private 子网具有 NAT 出口。

packages/infra/src/main.tf

module "my_database" {
  source              = "../../common/terraform/src/app/dbs/my-database"
  vpc_id              = aws_vpc.main.id
  database_subnet_ids = aws_subnet.database[*].id
  lambda_subnet_ids   = aws_subnet.private[*].id
}

module "api" {
  source     = "../../common/terraform/src/app/apis/my-api"
  enable_vpc = true
  vpc_id     = aws_vpc.main.id
  subnet_ids = aws_subnet.private[*].id

  appconfig_application_id  = module.runtime_config_appconfig.application_id
  appconfig_application_arn = module.runtime_config_appconfig.application_arn

  additional_iam_policy_statements = [
    {
      Effect = "Allow"
      Action = ["rds-db:connect"]
      Resource = [
        "arn:aws:rds-db:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:dbuser:${module.my_database.connect_resource_id}/${module.my_database.database_runtime_user}"
      ]
    }
  ]
}

resource "aws_vpc_security_group_ingress_rule" "api_to_database" {
  description                  = "Allow the API Lambda functions to connect to the database"
  security_group_id            = module.my_database.security_group_id
  referenced_security_group_id = module.api.security_group_id
  from_port                    = module.my_database.cluster_port
  to_port                      = module.my_database.cluster_port
  ip_protocol                  = "tcp"
}

resource "aws_vpc_security_group_egress_rule" "api_to_database" {
  description                  = "Allow outbound traffic from the API Lambda functions to the database"
  security_group_id            = module.api.security_group_id
  referenced_security_group_id = module.my_database.security_group_id
  from_port                    = module.my_database.cluster_port
  to_port                      = module.my_database.cluster_port
  ip_protocol                  = "tcp"
}

将 API Lambda 函数部署到具有出口的私有子网，而不是私有隔离子网。appconfig_application_id/appconfig_application_arn 来自在根模块中声明一次的共享运行时配置 AppConfig 应用程序，而不是来自数据库模块——传递它们会在 Lambda 函数上设置 RUNTIME_CONFIG_APP_ID 并授予它们对应用程序的读取访问权限。

AppConfig 应用程序必须公开 database 命名空间，以便部署数据库模块的运行时配置条目。在实例化 runtime-config/appconfig 模块时，将其包含在 namespaces 中：

packages/infra/src/main.tf

module "runtime_config_appconfig" {
  source = "../../common/terraform/src/core/runtime-config/appconfig"

  application_name = "my-app-runtime-config"
  namespaces       = ["connection", "agentcore", "database"]
}

本地开发

pnpm

pnpm nx dev <project-name>

yarn

yarn nx dev <project-name>

npm

npx nx dev <project-name>

bun

bunx nx dev <project-name>

这将启动 FastAPI 和所有连接的数据库。LOCAL_DEV=true 环境变量会使数据库客户端连接到其本地 Docker 数据库而不是 Aurora。