FastAPI

FastAPI 是一个用于在 Python 中构建 API 的框架。

FastAPI 生成器可创建带有 AWS CDK 或 Terraform 基础设施配置的新 FastAPI 项目。生成的后端使用 AWS Lambda 进行无服务器部署，通过 AWS API Gateway API 对外暴露。它配置了 AWS Lambda Powertools 用于可观测性，包括日志记录、AWS X-Ray 追踪和 Cloudwatch 指标。

使用方式

生成 FastAPI

您可以通过两种方式生成新的 FastAPI 项目：

VSCode
CLI

安装 Nx Console VSCode Plugin 如果您尚未安装
在VSCode中打开Nx控制台
点击 Generate (UI) 在"Common Nx Commands"部分
搜索 @aws/nx-plugin - py#fast-api
填写必需参数
点击 Generate

pnpm nx g @aws/nx-plugin:py#fast-api

yarn nx g @aws/nx-plugin:py#fast-api

npx nx g @aws/nx-plugin:py#fast-api

bunx nx g @aws/nx-plugin:py#fast-api

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

pnpm nx g @aws/nx-plugin:py#fast-api --dry-run

yarn nx g @aws/nx-plugin:py#fast-api --dry-run

npx nx g @aws/nx-plugin:py#fast-api --dry-run

bunx nx g @aws/nx-plugin:py#fast-api --dry-run

选项配置

参数	类型	默认值	描述
name 必需	string	-	Name of the API project to generate
computeType	string	ServerlessApiGatewayRestApi	The type of compute to use to deploy this API. Choose between ServerlessApiGatewayRestApi (default) or ServerlessApiGatewayHttpApi.
auth	string	IAM	The method used to authenticate with your API. Choose between IAM (default), Cognito or None.
directory	string	packages	The directory to store the application in.
iacProvider	string	CDK	The preferred IaC provider
moduleName	string	-	Python module name

生成器输出

生成器将在 <目录>/<API名称> 路径下创建以下项目结构：

project.json 项目配置和构建目标
pyproject.toml Python 项目配置和依赖项
文件夹<模块名称>
- __init__.py 模块初始化
- init.py 设置 FastAPI 应用并配置 powertools 中间件
- main.py API 实现
文件夹scripts
- generate_open_api.py 从 FastAPI 应用生成 OpenAPI 模式的脚本

基础设施

由于该生成器会根据您选择的 iacProvider 以基础设施即代码的形式输出，它将在 packages/common 目录下创建一个包含相关 CDK 构造体或 Terraform 模块的项目。

通用的基础设施即代码项目结构如下：

CDK
Terraform

文件夹packages/common/constructs
- 文件夹src
  - 文件夹app/ 针对特定项目/生成器的基础设施构造体
    …
  - 文件夹core/ 被 app 目录构造体重用的通用构造体
    …
  - index.ts 导出 app 目录构造体的入口文件
- project.json 项目构建目标与配置

文件夹packages/common/terraform
- 文件夹src
  - 文件夹app/ 针对特定项目/生成器的 Terraform 模块
    …
  - 文件夹core/ 被 app 目录模块重用的通用模块
    …
- project.json 项目构建目标与配置

部署 API 时会生成以下文件：

CDK
Terraform

文件夹packages/common/constructs/src
- 文件夹app
  - 文件夹apis
    <project-name>.ts 用于部署 API 的 CDK 构造
- 文件夹core
  - 文件夹api
    http-api.ts 部署 HTTP API 的 CDK 构造（如果你选择部署 HTTP API）
    rest-api.ts 部署 REST API 的 CDK 构造（如果你选择部署 REST API）
    utils.ts API 构造的实用工具

实现您的 FastAPI

主要 API 实现位于 main.py 中。这是定义 API 路由及其实现的地方。示例如下：

from .init import app, tracer
from pydantic import BaseModel

class Item(BaseModel):
  name: str

@app.get("/items/{item_id}")
def get_item(item_id: int) -> Item:
    return Item(name=...)

@app.post("/items")
def create_item(item: Item):
    return ...

生成器自动配置了以下功能：

用于可观测性的 AWS Lambda Powertools 集成
错误处理中间件
请求/响应关联
指标收集
使用 Mangum 的 AWS Lambda 处理程序

使用 AWS Lambda Powertools 进行可观测

日志记录

生成器使用 AWS Lambda Powertools 配置结构化日志记录。您可以在路由处理程序中访问日志记录器：

from .init import app, logger

@app.get("/items/{item_id}")
def read_item(item_id: int):
    logger.info("Fetching item", extra={"item_id": item_id})
    return {"item_id": item_id}

日志记录器自动包含：

用于请求追踪的关联 ID
请求路径和方法
Lambda 上下文信息
冷启动指示器

追踪

自动配置 AWS X-Ray 追踪。您可以为追踪添加自定义子段：

from .init import app, tracer

@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
    # 创建新的子段
    with tracer.provider.in_subsegment("fetch-item-details"):
        # 业务逻辑
        return {"item_id": item_id}

指标

自动收集每个请求的 CloudWatch 指标。您可以添加自定义指标：

from .init import app, metrics
from aws_lambda_powertools.metrics import MetricUnit

@app.get("/items/{item_id}")
def read_item(item_id: int):
    metrics.add_metric(name="ItemViewed", unit=MetricUnit.Count, value=1)
    return {"item_id": item_id}

默认指标包括：

请求计数
成功/失败计数
冷启动指标
按路由指标

错误处理

生成器包含全面的错误处理：

from fastapi import HTTPException

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id < 0:
        raise HTTPException(status_code=400, detail="Item ID must be positive")
    return {"item_id": item_id}

未捕获的异常会被中间件拦截并：

记录完整异常和堆栈跟踪
记录失败指标
向客户端返回安全的 500 响应
保留关联 ID

流式传输

使用 FastAPI 时，可以通过 StreamingResponse 响应类型向调用方流式传输响应。

基础设施变更

由于 AWS API Gateway 不支持流式响应，您需要将 FastAPI 部署到支持此功能的平台。最简单的选择是使用 AWS Lambda 函数 URL。

CDK
Terraform

要实现此功能，可以替换生成的 common/constructs/src/app/apis/<名称>-api.ts 构造，改用部署函数 URL 的构造。

流式 FunctionURL 构造示例

import { Duration, Stack, CfnOutput } from 'aws-cdk-lib';
import { IGrantable, Grant } from 'aws-cdk-lib/aws-iam';
import {
  Runtime,
  Code,
  Tracing,
  LayerVersion,
  FunctionUrlAuthType,
  InvokeMode,
  Function,
} from 'aws-cdk-lib/aws-lambda';
import { Construct } from 'constructs';
import url from 'url';
import { RuntimeConfig } from '../../core/runtime-config.js';

export class MyApi extends Construct {
  public readonly handler: Function;

  constructor(scope: Construct, id: string) {
    super(scope, id);

    this.handler = new Function(this, 'Handler', {
      runtime: Runtime.PYTHON_3_12,
      handler: 'run.sh',
      code: Code.fromAsset(
        url.fileURLToPath(
          new URL(
            '../../../../../../dist/packages/my_api/bundle',
            import.meta.url,
          ),
        ),
      ),
      timeout: Duration.seconds(30),
      tracing: Tracing.ACTIVE,
      environment: {
        AWS_CONNECTION_REUSE_ENABLED: '1',
      },
    });

    const stack = Stack.of(this);
    this.handler.addLayers(
      LayerVersion.fromLayerVersionArn(
        this,
        'LWALayer',
        `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
      ),
    );
    this.handler.addEnvironment('PORT', '8000');
    this.handler.addEnvironment('AWS_LWA_INVOKE_MODE', 'response_stream');
    this.handler.addEnvironment('AWS_LAMBDA_EXEC_WRAPPER', '/opt/bootstrap');
    const functionUrl = this.handler.addFunctionUrl({
      authType: FunctionUrlAuthType.AWS_IAM,
      invokeMode: InvokeMode.RESPONSE_STREAM,
      cors: {
        allowedOrigins: ['*'],
        allowedHeaders: [
          'authorization',
          'content-type',
          'x-amz-content-sha256',
          'x-amz-date',
          'x-amz-security-token',
        ],
      },
    });

    new CfnOutput(this, 'MyApiUrl', { value: functionUrl.url });

    // 在运行时配置中注册 API URL 供客户端发现
    RuntimeConfig.ensure(this).config.apis = {
      ...RuntimeConfig.ensure(this).config.apis!,
      MyApi: functionUrl.url,
    };
  }

  public grantInvokeAccess(grantee: IGrantable) {
    Grant.addToPrincipal({
      grantee,
      actions: ['lambda:InvokeFunctionUrl'],
      resourceArns: [this.handler.functionArn],
      conditions: {
        StringEquals: {
          'lambda:FunctionUrlAuthType': 'AWS_IAM',
        },
      },
    });
  }
}

要在 Terraform 中实现此功能，可以用支持响应流式传输的 Lambda 函数 URL 替换生成的 API Gateway 基础设施。

流式 Lambda 函数 URL 配置示例

# 当前 AWS 上下文的数据源
data "aws_caller_identity" "current" {}
data "aws_region" "current" {}

# 用于流式 FastAPI 的 Lambda 函数
resource "aws_lambda_function" "my_api_handler" {
  filename         = "../../../../../../dist/packages/my_api/bundle.zip"
  function_name    = "my-api-handler"
  role            = aws_iam_role.lambda_execution_role.arn
  handler         = "run.sh"
  runtime         = "python3.12"
  timeout         = 30
  source_code_hash = filebase64sha256("../../../../../../dist/packages/my_api/bundle.zip")

  # 启用 X-Ray 追踪
  tracing_config {
    mode = "Active"
  }

  # Lambda Web Adapter 的环境变量
  environment {
    variables = {
      AWS_CONNECTION_REUSE_ENABLED = "1"
      PORT                        = "8000"
      AWS_LWA_INVOKE_MODE        = "response_stream"
      AWS_LAMBDA_EXEC_WRAPPER    = "/opt/bootstrap"
    }
  }

  # 添加 Lambda Web Adapter 层
  layers = [
    "arn:aws:lambda:${data.aws_region.current.name}:753240598075:layer:LambdaAdapterLayerX86:24"
  ]

  depends_on = [
    aws_iam_role_policy_attachment.lambda_logs,
    aws_cloudwatch_log_group.lambda_logs,
  ]
}

# Lambda 函数的 CloudWatch 日志组
resource "aws_cloudwatch_log_group" "lambda_logs" {
  name              = "/aws/lambda/my-api-handler"
  retention_in_days = 14
}

# Lambda 执行 IAM 角色
resource "aws_iam_role" "lambda_execution_role" {
  name = "my-api-lambda-execution-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "lambda.amazonaws.com"
        }
      }
    ]
  })
}

# 附加基本执行策略
resource "aws_iam_role_policy_attachment" "lambda_logs" {
  role       = aws_iam_role.lambda_execution_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}

# 附加 X-Ray 追踪策略
resource "aws_iam_role_policy_attachment" "lambda_xray" {
  role       = aws_iam_role.lambda_execution_role.name
  policy_arn = "arn:aws:iam::aws:policy/AWSXRayDaemonWriteAccess"
}

# 支持流式传输的 Lambda 函数 URL
resource "aws_lambda_function_url" "my_api_url" {
  function_name      = aws_lambda_function.my_api_handler.function_name
  authorization_type = "AWS_IAM"
  invoke_mode       = "RESPONSE_STREAM"

  cors {
    allow_credentials = false
    allow_origins     = ["*"]
    allow_methods     = ["*"]
    allow_headers = [
      "authorization",
      "content-type",
      "x-amz-content-sha256",
      "x-amz-date",
      "x-amz-security-token"
    ]
    expose_headers = ["date", "keep-alive"]
    max_age       = 86400
  }
}

# 输出函数 URL
output "my_api_url" {
  description = "流式 FastAPI Lambda 函数的 URL"
  value       = aws_lambda_function_url.my_api_url.function_url
}

# 可选：为运行时配置创建 SSM 参数
resource "aws_ssm_parameter" "my_api_url" {
  name  = "/runtime-config/apis/MyApi"
  type  = "String"
  value = aws_lambda_function_url.my_api_url.function_url

  tags = {
    Environment = "production"
    Service     = "my-api"
  }
}

# 授予函数 URL 调用权限的 IAM 策略
resource "aws_iam_policy" "my_api_invoke_policy" {
  name        = "my-api-invoke-policy"
  description = "允许调用流式 FastAPI Lambda 函数 URL 的策略"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = "lambda:InvokeFunctionUrl"
        Resource = aws_lambda_function.my_api_handler.arn
        Condition = {
          StringEquals = {
            "lambda:FunctionUrlAuthType" = "AWS_IAM"
          }
        }
      }
    ]
  })
}

# 示例：将调用策略附加到角色（根据需要取消注释并修改）
# resource "aws_iam_role_policy_attachment" "my_api_invoke_access" {
#   role       = var.authenticated_role_name
#   policy_arn = aws_iam_policy.my_api_invoke_policy.arn
# }

实现

更新基础设施以支持流式传输后，可以在 FastAPI 中实现流式 API。API 应：

返回 StreamingResponse
声明每个响应块的返回类型
如果计划使用 API 连接，需添加 OpenAPI 供应商扩展 x-streaming: true

例如，如果要从 API 流式传输一系列 JSON 对象，可以按如下方式实现：

from pydantic import BaseModel
from fastapi.responses import StreamingResponse

class Chunk(BaseModel):
  message: str
  timestamp: datetime

async def stream_chunks():
  for i in range(0, 100):
    yield Chunk(message=f"This is chunk {i}", timestamp=datetime.now())

@app.get("/stream", openapi_extra={'x-streaming': True})
def my_stream() -> Chunk:
    return StreamingResponse(stream_chunks(), media_type="application/json")

消费

要消费流式响应，可以使用 API 连接生成器，该生成器将提供类型安全的方法来迭代流式数据块。

部署 FastAPI

FastAPI 生成器根据您选择的 iacProvider 创建 CDK 或 Terraform 基础设施即代码。您可以使用这些代码来部署 FastAPI。

CDK
Terraform

CDK 构造位于 common/constructs 文件夹中。您可以在 CDK 应用程序中使用：

import { MyApi } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // 将 API 添加到堆栈
    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
  }
}

此配置将设置：

为 FastAPI 应用中的每个操作创建 AWS Lambda 函数
作为函数触发器的 API Gateway HTTP/REST API
IAM 角色和权限
CloudWatch 日志组
X-Ray 追踪配置
CloudWatch 指标命名空间

如果选择使用 Cognito 身份验证，需要向 API 构造提供 identity 属性：

import { MyApi, UserIdentity } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const identity = new UserIdentity(this, 'Identity');

    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
      identity,
    });
  }
}

可以使用 ts#react-website-auth 生成器生成 UserIdentity 构造

Terraform 模块位于 common/terraform 文件夹中。您可以在 Terraform 配置中使用：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  # Lambda 函数的环境变量
  env = {
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }

  # 额外的 IAM 策略（如有需要）
  additional_iam_policy_statements = [
    # 添加 API 需要的任何额外权限
  ]

  tags = local.common_tags
}

此配置将设置：

服务所有 FastAPI 路由的 AWS Lambda 函数
作为函数触发器的 API Gateway HTTP/REST API
IAM 角色和权限
CloudWatch 日志组
X-Ray 追踪配置
CORS 配置

如果选择使用 Cognito 身份验证，需要提供 Cognito 配置：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  user_pool_id         = local.user_pool_id
  user_pool_client_ids = [local.client_id]

  env = {
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }

  tags = local.common_tags
}

您可以使用适当的 Terraform 资源或模块设置 Cognito 用户池和客户端。

Terraform 模块提供多个输出可供使用：

# 访问 API 端点
output "api_url" {
  value = module.my_api.stage_invoke_url
}

# 访问 Lambda 函数详情
output "lambda_function_name" {
  value = module.my_api.lambda_function_name
}

# 访问用于授予额外权限的 IAM 角色
output "lambda_execution_role_arn" {
  value = module.my_api.lambda_execution_role_arn
}

可以通过传递变量来自定义 CORS 设置：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  # 自定义 CORS 配置
  cors_allow_origins = ["https://myapp.com", "https://staging.myapp.com"]
  cors_allow_methods = ["GET", "POST", "PUT", "DELETE"]
  cors_allow_headers = [
    "authorization",
    "content-type",
    "x-custom-header"
  ]

  tags = local.common_tags
}

如果在运行生成器时选择 None 作为 auth 选项，可能会看到 Checkov 检查失败，例如：

HTTP API
REST API

Check: CKV_AWS_309: "Ensure API GatewayV2 routes specify an authorization type"
 FAILED for resource: aws_apigatewayv2_route.proxy_routes["PUT"]

Check: CKV_AWS_59: "Ensure there is no open access to back-end resources through API"
 FAILED for resource: aws_api_gateway_method.proxy_method

如果确定需要 API 公开访问，可以添加抑制注释

集成

REST/HTTP API 的 CDK 构造被配置为提供类型安全接口，用于为每个操作定义集成。

CDK
Terraform

CDK 构造提供如下所述完整的类型安全集成支持。

您可以使用静态方法 defaultIntegrations 来应用默认模式，这会为每个操作定义独立的 AWS Lambda 函数：

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this).build(),
});

Terraform 模块自动使用路由模式与单个 Lambda 函数，无需额外配置：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  # 模块自动创建处理所有 API 操作的
  # 单个 Lambda 函数
  tags = local.common_tags
}

访问集成

CDK
Terraform

您可以通过 API 构造的 integrations 属性以类型安全的方式访问底层 AWS Lambda 函数。例如，如果您的 API 定义了名为 sayHello 的操作，并需要为此函数添加权限：

const api = new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this).build(),
});

// sayHello 的类型与 API 中定义的操作相匹配
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [...],
  resources: [...],
}));

在 Terraform 的路由模式下，只有一个 Lambda 函数。可通过模块输出来访问：

# 为单个 Lambda 函数授予额外权限
resource "aws_iam_role_policy" "additional_permissions" {
  name = "additional-api-permissions"
  role = module.my_api.lambda_execution_role_name

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:GetObject",
          "s3:PutObject"
        ]
        Resource = "arn:aws:s3:::my-bucket/*"
      }
    ]
  })
}

自定义默认选项

CDK
Terraform

如需自定义创建 Lambda 函数时的默认选项，可使用 withDefaultOptions 方法。例如为所有 Lambda 函数配置 VPC：

const vpc = new Vpc(this, 'Vpc', ...);

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withDefaultOptions({
      vpc,
    })
    .build(),
});

要自定义 VPC 等配置，需要修改生成的 Terraform 模块。例如为所有 Lambda 函数添加 VPC 支持：

# 添加 VPC 变量
variable "vpc_subnet_ids" {
  description = "Lambda 函数的 VPC 子网 ID 列表"
  type        = list(string)
  default     = []
}

variable "vpc_security_group_ids" {
  description = "Lambda 函数的 VPC 安全组 ID 列表"
  type        = list(string)
  default     = []
}

# 更新 Lambda 函数资源
resource "aws_lambda_function" "api_lambda" {
  # ... 现有配置 ...

  # 添加 VPC 配置
  vpc_config {
    subnet_ids         = var.vpc_subnet_ids
    security_group_ids = var.vpc_security_group_ids
  }
}

使用带 VPC 配置的模块：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  # VPC 配置
  vpc_subnet_ids         = [aws_subnet.private_a.id, aws_subnet.private_b.id]
  vpc_security_group_ids = [aws_security_group.lambda_sg.id]

  tags = local.common_tags
}

覆盖集成

CDK
Terraform

使用 withOverrides 方法可以覆盖特定操作的集成。每个覆盖必须指定类型正确的 integration 属性。例如将 getDocumentation API 指向外部网站：

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getDocumentation: {
        integration: new HttpIntegration('https://example.com/documentation'),
      },
    })
    .build(),
});

覆盖后的集成通过 api.integrations.getDocumentation 访问时将不再具有 handler 属性。

您可以为集成添加额外属性以实现类型安全的抽象。例如为 REST API 创建 S3 集成后引用存储桶：

const storageBucket = new Bucket(this, 'Bucket', { ... });

const apiGatewayRole = new Role(this, 'ApiGatewayS3Role', {
  assumedBy: new ServicePrincipal('apigateway.amazonaws.com'),
});

storageBucket.grantRead(apiGatewayRole);

const api = new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getFile: {
        bucket: storageBucket,
        integration: new AwsIntegration({
          service: 's3',
          integrationHttpMethod: 'GET',
          path: `${storageBucket.bucketName}/{fileName}`,
          options: {
            credentialsRole: apiGatewayRole,
            requestParameters: {
              'integration.request.path.fileName': 'method.request.querystring.fileName',
            },
            integrationResponses: [{ statusCode: '200' }],
          },
        }),
        options: {
          requestParameters: {
            'method.request.querystring.fileName': true,
          },
          methodResponses: [{
            statusCode: '200',
          }],
        }
      },
    })
    .build(),
});

// 后续可类型安全地访问我们定义的 bucket 属性
api.integrations.getFile.bucket.grantRead(...);

覆盖授权器

CDK
Terraform

可在集成中提供 options 来覆盖特定方法选项，例如为 getDocumentation 操作使用 Cognito 认证：

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getDocumentation: {
        integration: new HttpIntegration('https://example.com/documentation'),
        options: {
          authorizer: new CognitoUserPoolsAuthorizer(...) // REST API 使用，HTTP API 使用 HttpUserPoolAuthorizer
        }
      },
    })
    .build(),
});

显式集成

CDK
Terraform

您可以选择不使用默认集成，直接为每个操作提供集成。这在需要不同集成类型时非常有用：

new MyApi(this, 'MyApi', {
  integrations: {
    sayHello: {
      integration: new LambdaIntegration(...),
    },
    getDocumentation: {
      integration: new HttpIntegration(...),
    },
  },
});

在 Terraform 中实现显式集成，需修改生成的模块：

移除默认代理路由（如 resource "aws_apigatewayv2_route" "proxy_routes"）
替换单个 Lambda 函数为各操作独立函数
为每个操作创建具体集成和路由，复用相同 ZIP 包：

HTTP API
REST API

# 移除默认 Lambda 函数和代理集成
 resource "aws_lambda_function" "api_lambda" { ... }
 resource "aws_apigatewayv2_integration" "lambda_integration" { ... }
 resource "aws_apigatewayv2_route" "proxy_routes" { ... }

# 添加各操作独立 Lambda 函数
 resource "aws_lambda_function" "say_hello_handler" {
   handler = "sayHello.handler"  # 特定操作处理器
   ...
 }

 resource "aws_lambda_function" "get_documentation_handler" {
   handler = "getDocumentation.handler"
   ...
 }

# 添加具体集成和路由
 resource "aws_apigatewayv2_integration" "say_hello_integration" { ... }
 resource "aws_apigatewayv2_route" "say_hello_route" { ... }
 resource "aws_apigatewayv2_integration" "get_documentation_integration" { ... }
 resource "aws_apigatewayv2_route" "get_documentation_route" { ... }

# 添加 Lambda 权限
 resource "aws_lambda_permission" "say_hello_permission" { ... }
 resource "aws_lambda_permission" "get_documentation_permission" { ... }

# 移除默认配置
 resource "aws_lambda_function" "api_lambda" { ... }
 resource "aws_apigatewayv2_integration" "lambda_integration" { ... }
 resource "aws_apigatewayv2_route" "proxy_routes" { ... }

# 添加各操作资源和集成
 resource "aws_api_gateway_resource" "say_hello_resource" { ... }
 resource "aws_api_gateway_method" "say_hello_method" { ... }
 resource "aws_api_gateway_integration" "say_hello_integration" { ... }

 resource "aws_api_gateway_resource" "get_documentation_resource" { ... }
 resource "aws_api_gateway_method" "get_documentation_method" { ... }
 resource "aws_api_gateway_integration" "get_documentation_integration" { ... }

# 更新部署依赖
~ resource "aws_api_gateway_deployment" "api_deployment" {
    depends_on = [
     aws_api_gateway_integration.say_hello_integration,
     aws_api_gateway_integration.get_documentation_integration,
    ]
  }

# 添加 Lambda 权限
 resource "aws_lambda_permission" "say_hello_permission" { ... }
 resource "aws_lambda_permission" "get_documentation_permission" { ... }

路由模式

CDK
Terraform

如需部署单个 Lambda 函数处理所有请求，可修改 defaultIntegrations 方法：

export class MyApi<...> extends ... {

  public static defaultIntegrations = (scope: Construct) => {
    const router = new Function(scope, 'RouterHandler', { ... });
    return IntegrationBuilder.rest({
      ...
      buildDefaultIntegration: (op) => {
        return {
          // 所有集成引用同一个路由处理器
          integration: new LambdaIntegration(router),
        };
      },
    });
  };
}

Terraform 模块默认使用路由模式，自动创建处理所有操作的单个 Lambda 函数：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  tags = local.common_tags
}

代码生成

CDK
Terraform

由于 FastAPI 的操作在 Python 中定义，而 CDK 基础设施在 TypeScript 中定义，我们通过代码生成机制向 CDK 构造提供元数据，以实现类型安全的集成接口。

在公共构造的 project.json 中添加了 generate:<ApiName>-metadata 目标以促进此代码生成，该目标会生成类似 packages/common/constructs/src/generated/my-api/metadata.gen.ts 的文件。由于这是在构建时生成的，因此版本控制中会忽略该文件。

每当更改 API 时都需要运行构建，以确保 CDK 构造使用的类型是最新的。

pnpm nx run-many --target build --all

yarn nx run-many --target build --all

npx nx run-many --target build --all

bunx nx run-many --target build --all

如果同时在进行 CDK 基础设施和 FastAPI 开发，可以使用 nx watch 在每次 API 更改时重新生成这些类型：

pnpm nx watch --projects=<FastAPIProject> -- \
pnpm nx run <InfraProject>:"generate:<ApiName>-metadata"

yarn nx watch --projects=<FastAPIProject> -- \
yarn nx run <InfraProject>:"generate:<ApiName>-metadata"

npx nx watch --projects=<FastAPIProject> -- \
npx nx run <InfraProject>:"generate:<ApiName>-metadata"

bunx nx watch --projects=<FastAPIProject> -- \
bunx nx run <InfraProject>:"generate:<ApiName>-metadata"

授予访问权限（仅限 IAM）

如果选择使用 IAM 身份验证，可以使用 grantInvokeAccess 方法授予 API 访问权限：

CDK
Terraform

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

# 创建允许调用 API 的 IAM 策略
resource "aws_iam_policy" "api_invoke_policy" {
  name        = "MyApiInvokePolicy"
  description = "允许调用 FastAPI 的策略"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = "execute-api:Invoke"
        Resource = "${module.my_api.api_execution_arn}/*/*"
      }
    ]
  })
}

# 将策略附加到 IAM 角色（例如已验证用户角色）
resource "aws_iam_role_policy_attachment" "api_invoke_access" {
  role       = aws_iam_role.authenticated_user_role.name
  policy_arn = aws_iam_policy.api_invoke_policy.arn
}

# 或按名称附加到现有角色
resource "aws_iam_role_policy_attachment" "api_invoke_access_existing" {
  role       = "MyExistingRole"
  policy_arn = aws_iam_policy.api_invoke_policy.arn
}

API 模块的关键输出可用于 IAM 策略：

module.my_api.api_execution_arn - 用于授予 execute-api:Invoke 权限
module.my_api.api_arn - API Gateway ARN
module.my_api.lambda_function_arn - Lambda 函数 ARN

本地开发

生成器配置了本地开发服务器，可通过以下命令运行：

pnpm nx run my-api:serve

yarn nx run my-api:serve

npx nx run my-api:serve

bunx nx run my-api:serve

这将启动本地 FastAPI 开发服务器，包含：

代码更改时自动重载
在 /docs 或 /redoc 提供交互式 API 文档
在 /openapi.json 提供 OpenAPI 模式

调用 FastAPI

要从 React 网站调用 API，可以使用 api-connection 生成器。