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.
integrationPattern	string	isolated	How API Gateway integrations are generated for the API. Choose between isolated (default) and shared.
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	Inherit	The preferred IaC provider. By default this is inherited from your initial selection.
moduleName	string	-	Python module name

生成器输出

生成器将在 <directory>/<api-name> 目录下创建以下项目结构:

project.json 项目配置和构建目标
pyproject.toml Python 项目配置和依赖项
run.sh Lambda Web Adapter 引导脚本,通过 uvicorn 启动 FastAPI 应用
文件夹<module_name>
- __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 pydantic import BaseModel
from .init import app, tracer

class Item(BaseModel):
  name: str

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

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

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

用于可观测性的 AWS Lambda Powertools 集成
错误处理中间件
请求/响应关联
指标收集
通过 Lambda Web Adapter 使用 uvicorn 进行 AWS Lambda 部署
类型安全的流式传输(仅限 REST API)

使用 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 在使用 REST API 时开箱即支持流式响应。基础设施配置为使用 AWS Lambda Web Adapter 在 Lambda 内通过 uvicorn 运行您的 FastAPI,并在 API Gateway 中为所有 REST API 操作设置 ResponseTransferMode.STREAM,这使得流式传输可以与非流式操作并行工作。

使用 `JsonStreamingResponse`

生成的 init.py 导出了 JsonStreamingResponse 类,该类提供类型安全的流式传输和正确的 OpenAPI 模式生成。这确保 connection 生成器可以生成正确类型的流式客户端方法。

from pydantic import BaseModel
from .init import app, JsonStreamingResponse

class Chunk(BaseModel):
    message: str

async def generate_chunks():
    for i in range(100):
        yield Chunk(message=f"This is chunk {i}")

@app.post(
    "/stream",
    response_class=JsonStreamingResponse,
    responses={200: JsonStreamingResponse.openapi_response(Chunk, "Stream of chunks")},
)
async def my_stream() -> JsonStreamingResponse:
    return JsonStreamingResponse(generate_chunks())

JsonStreamingResponse 类:

将 Pydantic 模型序列化为 JSON Lines 格式(application/jsonl)
提供 openapi_response 辅助方法,生成包含 itemSchema 的正确 OpenAPI 模式,使 connection 生成器能够生成类型安全的流式客户端方法

消费

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

部署 FastAPI

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

CDK
Terraform

用于部署 API 的 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 指标命名空间

如果您的解决方案包含网站，您可以将其 CloudFront 分发配置为 API 网关 / API AWS Lambda 集成中 HTTP / REST API 唯一允许的 CORS 源。您可以使用网站构造、CloudFront 分发、来源字符串或它们的组合来调用 API 的 restrictCorsTo 方法。

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
    const website = new MyWebsite(this, 'MyWebsite');
    api.restrictCorsTo(website, 'http://localhost:4200');
  }
}

MyWebsite 构造可以使用 ts#react-website generator 生成

如果选择使用 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 构造

用于部署 API 的 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 配置

如果您的解决方案包含 Terraform 网站模块，那么您可以使用其 CloudFront 域名来限制 CORS。给定一个 CloudFront 域名 <domain_name>，在用于部署 API 的 Terraform 模块中添加

对于 HTTP API，添加一个 cors_allow_origins 属性，设置为 ["http://localhost:4200", "http://localhost:4300", "https://<domain name>"]。这会将 API 网关 CORS 限制为此分发和本地主机。
对于 REST API，添加一个 ALLOWED_ORIGINS 环境变量，设置为 "https://<domain_name>"。这会将 CloudFront 分发设置为 AWS Lambda 集成中唯一允许的 CORS 源（除本地主机外）。请注意，此限制不适用于预检 OPTIONS 请求 - 请为 this GitHub issue 点赞以帮助优先解决此问题。

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

  cors_allow_origins = ["http://localhost:4200", "http://localhost:4300", "https://<domain name>"] // Only required for HTTP API

  env = {
    ALLOWED_ORIGINS = "https://<domain name>" // Only required for REST API
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }
}

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

如果选择使用 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: [...],
}));

如果您的 API 使用 shared 模式，共享路由 Lambda 函数通过 api.integrations.$router 暴露：

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

api.integrations.$router.handler.addEnvironment('LOG_LEVEL', 'DEBUG');

在 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(),
});

如果您在默认选项中指定 environment，这将__覆盖__在生成的 API 构造中定义的任何环境变量。通常更推荐使用 Lambda 函数的 addEnvironment 方法。您可以通过循环遍历集成来轻松实现：

Object.values(api.integrations).forEach(({ handler }) => {
  handler.addEnvironment('LOG_LEVEL', 'INFO');
});

要自定义 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 属性，其类型对应 HTTP 或 REST API 的相应 CDK 集成构造。withOverrides 方法也是类型安全的。例如将 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 中实现显式集成，需修改生成的应用特定模块，将默认代理集成替换为每个操作的特定集成。

编辑 packages/common/terraform/src/app/apis/my-api/my-api.tf：

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

HTTP API
REST API

# 移除默认单个 Lambda 函数
 resource "aws_lambda_function" "api_lambda" {
   filename         = data.archive_file.lambda_zip.output_path
   function_name    = "MyApiHandler"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "index.handler"
   runtime         = "nodejs22.x"
   timeout         = 30
   # ... 其余配置
 }

# 移除默认代理集成
 resource "aws_apigatewayv2_integration" "lambda_integration" {
   api_id           = module.http_api.api_id
   integration_type = "AWS_PROXY"
   integration_uri  = aws_lambda_function.api_lambda.invoke_arn
   # ... 其余配置
 }

# 移除默认代理路由
 resource "aws_apigatewayv2_route" "proxy_routes" {
   for_each = toset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"])
   api_id    = module.http_api.api_id
   route_key = "${each.key} /{proxy+}"
   target    = "integrations/${aws_apigatewayv2_integration.lambda_integration.id}"
   # ... 其余配置
 }

# 为每个操作添加独立 Lambda 函数，使用相同的包
 resource "aws_lambda_function" "say_hello_handler" {
   filename         = data.archive_file.lambda_zip.output_path
   function_name    = "MyApi-SayHello"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "sayHello.handler"  # 此操作的特定处理器
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = merge({
       AWS_CONNECTION_REUSE_ENABLED = "1"
     }, var.env)
   }

   tags = var.tags
 }

 resource "aws_lambda_function" "get_documentation_handler" {
   filename         = data.archive_file.lambda_zip.output_path
   function_name    = "MyApi-GetDocumentation"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "getDocumentation.handler"  # 此操作的特定处理器
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = merge({
       AWS_CONNECTION_REUSE_ENABLED = "1"
     }, var.env)
   }

   tags = var.tags
 }

# 为每个操作添加特定集成
 resource "aws_apigatewayv2_integration" "say_hello_integration" {
   api_id           = module.http_api.api_id
   integration_type = "AWS_PROXY"
   integration_uri  = aws_lambda_function.say_hello_handler.invoke_arn
   payload_format_version = "2.0"
   timeout_milliseconds   = 30000
 }

 resource "aws_apigatewayv2_integration" "get_documentation_integration" {
   api_id           = module.http_api.api_id
   integration_type = "HTTP_PROXY"
   integration_uri  = "https://example.com/documentation"
   integration_method = "GET"
 }

# 为每个操作添加特定路由
 resource "aws_apigatewayv2_route" "say_hello_route" {
   api_id    = module.http_api.api_id
   route_key = "POST /sayHello"
   target    = "integrations/${aws_apigatewayv2_integration.say_hello_integration.id}"
   authorization_type = "AWS_IAM"
 }

 resource "aws_apigatewayv2_route" "get_documentation_route" {
   api_id    = module.http_api.api_id
   route_key = "GET /documentation"
   target    = "integrations/${aws_apigatewayv2_integration.get_documentation_integration.id}"
   authorization_type = "NONE"
 }

# 为每个函数添加 Lambda 权限
 resource "aws_lambda_permission" "say_hello_permission" {
   statement_id  = "AllowExecutionFromAPIGateway-SayHello"
   action        = "lambda:InvokeFunction"
   function_name = aws_lambda_function.say_hello_handler.function_name
   principal     = "apigateway.amazonaws.com"
   source_arn    = "${module.http_api.api_execution_arn}/*/*"
 }

 resource "aws_lambda_permission" "get_documentation_permission" {
   statement_id  = "AllowExecutionFromAPIGateway-GetDocumentation"
   action        = "lambda:InvokeFunction"
   function_name = aws_lambda_function.get_documentation_handler.function_name
   principal     = "apigateway.amazonaws.com"
   source_arn    = "${module.http_api.api_execution_arn}/*/*"
 }

# 移除默认单个 Lambda 函数
 resource "aws_lambda_function" "api_lambda" {
   filename         = data.archive_file.lambda_zip.output_path
   function_name    = "MyApiHandler"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "index.handler"
   runtime         = "nodejs22.x"
   timeout         = 30
   # ... 其余配置
 }

# 移除默认代理集成
 resource "aws_apigatewayv2_integration" "lambda_integration" {
   api_id           = module.http_api.api_id
   integration_type = "AWS_PROXY"
   integration_uri  = aws_lambda_function.api_lambda.invoke_arn
   # ... 其余配置
 }

# 移除默认代理路由
 resource "aws_apigatewayv2_route" "proxy_routes" {
   for_each = toset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"])
   api_id    = module.http_api.api_id
   route_key = "${each.key} /{proxy+}"
   target    = "integrations/${aws_apigatewayv2_integration.lambda_integration.id}"
   # ... 其余配置
 }

# 为每个操作添加独立 Lambda 函数，使用相同的包
 resource "aws_lambda_function" "say_hello_handler" {
   filename         = data.archive_file.lambda_zip.output_path
   function_name    = "MyApi-SayHello"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "sayHello.handler"  # 此操作的特定处理器
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = merge({
       AWS_CONNECTION_REUSE_ENABLED = "1"
     }, var.env)
   }

   tags = var.tags
 }

 resource "aws_lambda_function" "get_documentation_handler" {
   filename         = data.archive_file.lambda_zip.output_path
   function_name    = "MyApi-GetDocumentation"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "getDocumentation.handler"  # 此操作的特定处理器
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = merge({
       AWS_CONNECTION_REUSE_ENABLED = "1"
     }, var.env)
   }

   tags = var.tags
 }

# 为每个操作添加特定资源和方法
 resource "aws_api_gateway_resource" "say_hello_resource" {
   rest_api_id = module.rest_api.api_id
   parent_id   = module.rest_api.api_root_resource_id
   path_part   = "sayHello"
 }

 resource "aws_api_gateway_method" "say_hello_method" {
   rest_api_id   = module.rest_api.api_id
   resource_id   = aws_api_gateway_resource.say_hello_resource.id
   http_method   = "POST"
   authorization = "AWS_IAM"
 }

 resource "aws_api_gateway_integration" "say_hello_integration" {
   rest_api_id = module.rest_api.api_id
   resource_id = aws_api_gateway_resource.say_hello_resource.id
   http_method = aws_api_gateway_method.say_hello_method.http_method

   integration_http_method = "POST"
   type                   = "AWS_PROXY"
   uri                    = aws_lambda_function.say_hello_handler.invoke_arn
 }

 resource "aws_api_gateway_resource" "get_documentation_resource" {
   rest_api_id = module.rest_api.api_id
   parent_id   = module.rest_api.api_root_resource_id
   path_part   = "documentation"
 }

 resource "aws_api_gateway_method" "get_documentation_method" {
   rest_api_id   = module.rest_api.api_id
   resource_id   = aws_api_gateway_resource.get_documentation_resource.id
   http_method   = "GET"
   authorization = "NONE"
 }

 resource "aws_api_gateway_integration" "get_documentation_integration" {
   rest_api_id = module.rest_api.api_id
   resource_id = aws_api_gateway_resource.get_documentation_resource.id
   http_method = aws_api_gateway_method.get_documentation_method.http_method

   integration_http_method = "GET"
   type                   = "HTTP"
   uri                    = "https://example.com/documentation"
 }

# 更新部署以依赖新集成
~ resource "aws_api_gateway_deployment" "api_deployment" {
    rest_api_id = module.rest_api.api_id

    depends_on = [
     aws_api_gateway_integration.lambda_integration,
     aws_api_gateway_integration.say_hello_integration,
     aws_api_gateway_integration.get_documentation_integration,
    ]

    lifecycle {
      create_before_destroy = true
    }

   triggers = {
     redeployment = sha1(jsonencode([
       aws_api_gateway_integration.say_hello_integration,
       aws_api_gateway_integration.get_documentation_integration,
     ]))
   }
  }

# 为每个函数添加 Lambda 权限
 resource "aws_lambda_permission" "say_hello_permission" {
   statement_id  = "AllowExecutionFromAPIGateway-SayHello"
   action        = "lambda:InvokeFunction"
   function_name = aws_lambda_function.say_hello_handler.function_name
   principal     = "apigateway.amazonaws.com"
   source_arn    = "${module.rest_api.api_execution_arn}/*/*"
 }

 resource "aws_lambda_permission" "get_documentation_permission" {
   statement_id  = "AllowExecutionFromAPIGateway-GetDocumentation"
   action        = "lambda:InvokeFunction"
   function_name = aws_lambda_function.get_documentation_handler.function_name
   principal     = "apigateway.amazonaws.com"
   source_arn    = "${module.rest_api.api_execution_arn}/*/*"
 }

集成模式

CDK
Terraform

生成的 CDK API 构造支持两种集成模式：

isolated 为每个操作创建一个 Lambda 函数。这是生成 API 的默认模式。
shared 创建单个默认路由 Lambda 函数并在所有操作中复用，除非您覆盖特定集成。

isolated 为每个操作提供更细粒度的权限和配置。shared 减少 Lambda 和 API Gateway 集成的数量，同时仍允许选择性覆盖。

例如，将 pattern 设置为 'shared' 会创建单个函数而不是为每个集成创建一个：

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

  public static defaultIntegrations = (scope: Construct) => {
    ...
    return IntegrationBuilder.rest({
      pattern: 'shared',
      ...
    });
  };
}

Terraform 模块自动使用路由模式 - 这是默认且唯一支持的方式。生成的模块创建单个 Lambda 函数处理所有 API 操作。

您只需实例化默认模块即可获得路由模式：

# 默认路由模式 - 单个 Lambda 函数处理所有操作
module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  # 单个 Lambda 函数自动处理所有操作
  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 build

yarn build

npm run build

bun build

如果同时进行 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,可以使用 connection 生成器。