FastAPI

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

FastAPI 生成器创建一个新的 FastAPI，并配置 AWS CDK 或 Terraform 基础设施。生成的后端使用 AWS Lambda 进行无服务器部署，通过 AWS API Gateway API 公开。它设置了 AWS Lambda Powertools 以实现可观测性，包括日志记录、AWS X-Ray 追踪和 Cloudwatch 指标。

用法

生成 FastAPI

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

CLI
VSCode

pnpm nx g @aws/nx-plugin:py#api --framework=fastapi

yarn nx g @aws/nx-plugin:py#api --framework=fastapi

npx nx g @aws/nx-plugin:py#api --framework=fastapi

bunx nx g @aws/nx-plugin:py#api --framework=fastapi

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

pnpm nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

yarn nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

npx nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

bunx nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

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

选项

参数	类型	默认值	描述
name 必需	string	-	要生成的 API 项目名称
framework	fastapi	fastapi	要使用的API框架。
integrationPattern	isolated \| shared	isolated	为 API 生成 API Gateway 集成的方式。可选择 isolated（默认）或 shared。
auth	iam \| cognito \| custom	iam	用于对 API 进行身份验证的方法。可选择 iam（默认）、cognito 或 custom。
directory	string	packages	存储应用程序的目录。
subDirectory	string	-	项目所在的子目录。默认情况下为项目名称。
iac	inherit \| cdk \| terraform	inherit	首选的 IaC 提供商。默认情况下，这将继承您的初始选择。
moduleName	string	-	Python 模块名称
infra	rest-lambda \| http-lambda \| none	rest-lambda	用于部署此 API 的基础设施类型。
preferInstallDependencies	boolean	true	是否在生成器运行后优先安装依赖项。设置为 false 可在批量运行多个生成器时延迟安装（如果后续生成器需要计算 Nx 项目图，仍会运行安装）；在最后统一安装一次。

生成器输出

生成器将在 <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 架构的脚本

基础设施

由于此生成器根据您选择的 iac 提供基础设施即代码，它将在 packages/common 中创建一个项目，其中包含相关的 CDK 构造或 Terraform 模块。

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

CDK
Terraform

文件夹packages/common/constructs
- 文件夹src
  - 文件夹app/ Constructs for infrastructure specific to a project/generator
    …
  - 文件夹core/ Generic constructs which are reused by constructs in app
    …
  - index.ts Entry point exporting constructs from app
- project.json Project build targets and configuration

文件夹packages/common/terraform
- 文件夹src
  - 文件夹app/ Terraform modules for infrastructure specific to a project/generator
    …
  - 文件夹core/ Generic modules which are reused by modules in app
    …
- project.json Project build targets and configuration

为了部署您的 API，会生成以下文件：

CDK
Terraform

文件夹packages/common/constructs/src
- 文件夹app
  - 文件夹apis
    <project-name>.ts CDK construct for deploying your API
- 文件夹core
  - 文件夹api
    http-api.ts CDK construct for deploying an HTTP API (if you selected to deploy an HTTP API)
    rest-api.ts CDK construct for deploying a REST API (if you selected to deploy a REST API)
    utils.ts Utilities for the API constructs

架构

已部署的应用程序具有以下架构：

REST API
HTTP API

REST API 在 API Gateway 阶段前包含一个 AWS WAFv2 Web ACL，并启用了 AWS 托管的默认规则集。

实现您的 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
请求路径、匹配的路由和方法

追踪

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

from .init import app, tracer

@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("fetch-item-details"):
        # Your logic here
        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}

默认指标包括：

请求计数
成功/失败计数
按路由的指标（通过 <method> <path> 的 route 维度）

错误处理

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

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

访问调用用户

当您的 API 受身份验证保护时，您的路由处理程序通常需要知道是谁在调用。生成的 FastAPI 通过 Lambda Web Adapter 在 AWS Lambda 内部运行，它将 API Gateway 请求上下文作为 JSON 转发到 x-amzn-request-context 标头。您可以从 FastAPI Request 中读取它以提取调用者的身份。

例如，让我们添加一个 /me 端点，返回有关调用用户的详细信息。我们将提取实现为 FastAPI 依赖项，以便可以在路由之间重用。请求上下文的形状——因此您如何提取身份——取决于您选择的 auth 方法以及您是否部署了 REST 或 HTTP API。

对于 IAM 身份验证，我们使用从 API Gateway 请求上下文中提取的 sub 在 Cognito 中查找调用者。在 main.py 旁边创建 identity.py：

REST API
HTTP API

import json
import os
from typing import Annotated

from boto3 import client
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel

cognito = client("cognito-idp")


class Identity(BaseModel):
    sub: str
    username: str


def get_identity(request: Request) -> Identity:
    # The Lambda Web Adapter forwards the API Gateway request context as JSON
    request_context_header = request.headers.get("x-amzn-request-context")
    if not request_context_header:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    request_context = json.loads(request_context_header)
    provider = request_context.get("identity", {}).get("cognitoAuthenticationProvider")

    sub = provider.split(":")[-1] if provider else None
    if not sub:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    users = cognito.list_users(
        # Assumes user pool id is configured in lambda environment
        UserPoolId=os.environ["USER_POOL_ID"],
        Limit=1,
        Filter=f'sub="{sub}"',
    ).get("Users", [])

    if len(users) != 1:
        raise HTTPException(status_code=403, detail=f"No user found with subjectId {sub}")

    return Identity(sub=sub, username=users[0]["Username"])


CurrentUser = Annotated[Identity, Depends(get_identity)]

import json
import os
from typing import Annotated

from boto3 import client
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel

cognito = client("cognito-idp")


class Identity(BaseModel):
    sub: str
    username: str


def get_identity(request: Request) -> Identity:
    # The Lambda Web Adapter forwards the API Gateway request context as JSON
    request_context_header = request.headers.get("x-amzn-request-context")
    if not request_context_header:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    request_context = json.loads(request_context_header)
    amr = (
        request_context.get("authorizer", {})
        .get("iam", {})
        .get("cognitoIdentity", {})
        .get("amr", [])
    )
    sign_in = next((s for s in amr if ":CognitoSignIn:" in s), None)
    sub = sign_in.split(":")[-1] if sign_in else None

    if not sub:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    users = cognito.list_users(
        # Assumes user pool id is configured in lambda environment
        UserPoolId=os.environ["USER_POOL_ID"],
        Limit=1,
        Filter=f'sub="{sub}"',
    ).get("Users", [])

    if len(users) != 1:
        raise HTTPException(status_code=403, detail=f"No user found with subjectId {sub}")

    return Identity(sub=sub, username=users[0]["Username"])


CurrentUser = Annotated[Identity, Depends(get_identity)]

使用 auth: 'cognito'，API Gateway Cognito User Pools 授权器验证调用者在 Authorization 标头中提供的 JWT，并将验证的声明放在请求上下文中。

在 main.py 旁边创建 identity.py：

REST API
HTTP API

import json
from typing import Annotated

from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel


class Identity(BaseModel):
    sub: str
    username: str


def get_identity(request: Request) -> Identity:
    # The Lambda Web Adapter forwards the API Gateway request context as JSON
    request_context_header = request.headers.get("x-amzn-request-context")
    if not request_context_header:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    request_context = json.loads(request_context_header)
    claims = request_context.get("authorizer", {}).get("claims", {})

    sub = claims.get("sub")
    username = claims.get("username")

    if not sub or not username:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    return Identity(sub=sub, username=username)


CurrentUser = Annotated[Identity, Depends(get_identity)]

HTTP API 使用 JWT 授权器，它将验证的声明放在 authorizer.jwt.claims 下：

import json
from typing import Annotated

from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel


class Identity(BaseModel):
    sub: str
    username: str


def get_identity(request: Request) -> Identity:
    # The Lambda Web Adapter forwards the API Gateway request context as JSON
    request_context_header = request.headers.get("x-amzn-request-context")
    if not request_context_header:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    request_context = json.loads(request_context_header)
    claims = request_context.get("authorizer", {}).get("jwt", {}).get("claims", {})

    sub = claims.get("sub")
    username = claims.get("username")

    if not sub or not username:
        raise HTTPException(status_code=403, detail="Unable to determine calling user")

    return Identity(sub=sub, username=username)


CurrentUser = Annotated[Identity, Depends(get_identity)]

然后，您可以将 CurrentUser 依赖项注入到任何需要调用者身份的路由中：

from .identity import CurrentUser, Identity
from .init import app, tracer

@app.get("/me")
@tracer.capture_method
def me(identity: CurrentUser) -> Identity:
    return identity

流式传输

使用 REST API 时，生成的 FastAPI 开箱即支持流式响应。基础设施配置为使用 AWS Lambda Web Adapter 通过 uvicorn 在 Lambda 内部运行您的 FastAPI，并为所有 REST API 操作在 API Gateway 中使用 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 生成器根据您选择的 iac 创建 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) {
    // Add the api to your stack
    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#website 生成器生成

如果您选择使用 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,
    });
  }
}

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

用于部署 API 的 Terraform 模块在 common/terraform 文件夹中。您可以在 Terraform 配置中使用它。

API 模块将其 Lambda 部署 zip 暂存在共享的 S3 资产存储桶中——有关详细信息，请参阅 Terraform 基础设施指南。每次部署实例化一次 core/asset-bucket 模块，并通过 asset_bucket_name 输入将其 bucket_name 输出传递到每个 API / Lambda 模块：

module "asset_bucket" {
  source = "../../common/terraform/src/core/asset-bucket"
}

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

  asset_bucket_name = module.asset_bucket.bucket_name

  # Environment variables for the Lambda function
  env = {
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }

  # Additional IAM policies if needed
  additional_iam_policy_statements = [
    # Add any additional permissions your API needs
  ]

  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 - 请为此 GitHub issue 点赞以帮助优先解决此问题。

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

  asset_bucket_name = module.asset_bucket.bucket_name

  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#website 生成器生成

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

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

  asset_bucket_name    = module.asset_bucket.bucket_name

  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 User Pool 和 Client。

Terraform 模块提供了几个您可以使用的输出：

# Access the API endpoint
output "api_url" {
  value = module.my_api.stage_invoke_url
}

# Access Lambda function details
output "lambda_function_name" {
  value = module.my_api.lambda_function_name
}

# Access IAM role for granting additional permissions
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"

  asset_bucket_name = module.asset_bucket.bucket_name

  # Custom CORS configuration
  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
}

WAF

对于 REST API，生成的构造默认会将 AWS WAFv2 Web ACL 与 API Gateway 阶段关联。Web ACL 使用 AWS 托管的默认规则集（AWSManagedRulesCommonRuleSet 和 AWSManagedRulesKnownBadInputsRuleSet），提供针对常见 Web 漏洞（包括 OWASP Top 10）的保护。WAF 请求日志会写入 CloudWatch Logs 日志组。

您可以编辑生成的 rest-api 构造来添加、删除或调整规则（例如，添加基于速率的规则或其他托管规则组）。

CDK
Terraform

要选择退出（例如，附加您自己的 Web ACL），请将 enableWaf 设置为 false：

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

要选择退出（例如，附加您自己的 Web ACL），请将 enable_waf 设置为 false：

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

  asset_bucket_name = module.asset_bucket.bucket_name
  enable_waf        = false
}

访问日志记录

对于 REST API，生成的基础设施默认启用访问日志，为每个请求向专用的 CloudWatch Logs 日志组写入一行结构化的 JSON。该日志组使用客户管理的 KMS 密钥加密，并保留一年。

API Gateway 使用账户级别的 CloudWatch Logs 角色写入访问日志。此角色在 AWS::ApiGateway::Account 设置上配置，该设置是每个区域每个账户的单例 — 区域中的每个 REST API 只有一个角色。为了在多个独立部署的堆栈中安全地管理此角色，生成的基础设施：

仅在尚未设置有效角色时创建共享的 CloudWatch Logs 角色并在账户上配置它，因此部署永远不会覆盖另一个堆栈拥有的角色。
在拆除时保持账户设置不变，因此销毁一个堆栈永远不会禁用区域中其他 REST API 的日志记录。

CDK
Terraform

账户角色由 ApiGatewayAccount 构造管理，这是一个通过 ApiGatewayAccount.ensure(scope) 解析的堆栈范围单例。每个 REST API 的阶段都依赖于它，该角色由 Lambda 支持的自定义资源配置。

访问日志格式由您的 API 扩展的 RestApi 构造设置。要自定义它，请在生成的 packages/common/constructs/src/app/apis/my-api.ts 中将 deployOptions 传递给 super，同时保留构造已设置的 tracingEnabled：

super(scope, id, {
  apiName: 'MyApi',
  // ...
  deployOptions: {
    tracingEnabled: true,
    accessLogFormat: AccessLogFormat.clf(),
  },
  ...props,
});

AccessLogFormat 从 aws-cdk-lib/aws-apigateway 导入。任何未设置的内容都会保留构造的默认值 — 一个包含标准字段的 JSON 格式。

账户角色由 core/api/api-gateway-account 模块管理，该模块由生成的 API 模块实例化。它以幂等方式配置账户，并且在 terraform destroy 时永远不会重置。

您可以通过编辑生成的 API 模块中 aws_api_gateway_stage 资源上的 access_log_settings 块来自定义访问日志格式。

集成

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"

  asset_bucket_name = module.asset_bucket.bucket_name

  # The module automatically creates a single Lambda function
  # that handles all API operations
  tags = local.common_tags
}

访问集成

CDK
Terraform

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

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

// sayHello is typed to the operations defined in your 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 函数。您可以通过模块输出访问它：

# Grant additional permissions to the single Lambda function
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 配置已由生成的模块支持 — 设置 enable_vpc 以及 vpc_id 和 subnet_ids，模块会将 Lambda 函数部署到您的 VPC 中，并为您创建一个安全组：

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

  asset_bucket_name = module.asset_bucket.bucket_name

  # VPC configuration
  enable_vpc = true
  vpc_id     = aws_vpc.main.id
  subnet_ids = [aws_subnet.private_a.id, aws_subnet.private_b.id]

  tags = local.common_tags
}

对于模块未公开的选项，请直接编辑生成的 Terraform 模块中的 aws_lambda_function 资源。

按操作自定义选项

CDK
Terraform

要自定义用于创建_特定_操作的默认集成的选项（而不影响其他操作），可以使用 withOperationOptions 方法。例如，如果您想仅为一个操作增加 Lambda 函数超时时间：

const api = new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOperationOptions({
      sayHello: {
        timeout: Duration.seconds(60),
      },
    })
    .build(),
});

// The selected operations remain default integrations, so they're still typed accordingly:
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

您指定的选项将与默认集成选项（以及通过 withDefaultOptions 设置的任何选项）合并。请注意，您不能为通过 withOverrides 替换的操作指定选项，因为这些操作不再使用默认集成。

如果同一操作同时被 withOperationOptions 和 withOverrides 针对，无论您调用它们的顺序如何，都会遇到类型错误。

覆盖集成

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

// Later, perhaps in another file, you can access the bucket property we defined
// in a type-safe manner
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(...) // for REST, or HttpUserPoolAuthorizer for an HTTP API
        }
      },
    })
    .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

# Remove the default single Lambda function
 resource "aws_lambda_function" "api_lambda" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApiHandler"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "index.handler"
   runtime         = "nodejs22.x"
   timeout         = 30
   # ... rest of configuration
 }

# Remove the default proxy integration
 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
   # ... rest of configuration
 }

# Remove the default proxy routes
 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}"
   # ... rest of configuration
 }

# Add individual Lambda functions for each operation using the same bundle
 resource "aws_lambda_function" "say_hello_handler" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-SayHello"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "sayHello.handler"  # Specific handler for this operation
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = var.env
   }

   tags = var.tags
 }

 resource "aws_lambda_function" "get_documentation_handler" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-GetDocumentation"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "getDocumentation.handler"  # Specific handler for this operation
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = var.env
   }

   tags = var.tags
 }

# Add specific integrations for each operation
 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"
 }

# Add specific routes for each operation
 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"
 }

# Add Lambda permissions for each function
 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}/*/*"
 }

# Remove the default single Lambda function
 resource "aws_lambda_function" "api_lambda" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApiHandler-${random_string.suffix.result}"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "index.handler"
   runtime         = "nodejs22.x"
   timeout         = 30
   # ... rest of configuration
 }

# Remove the default proxy integration
 resource "aws_api_gateway_integration" "lambda_integration" {
   rest_api_id = module.rest_api.api_id
   resource_id = aws_api_gateway_resource.proxy_resource.id
   http_method = aws_api_gateway_method.proxy_method.http_method
   integration_http_method = "POST"
   type                    = "AWS_PROXY"
   uri                     = aws_lambda_function.api_lambda.invoke_arn
   # ... rest of configuration
 }

# Remove the default catch-all proxy method
 resource "aws_api_gateway_method" "proxy_method" {
   rest_api_id   = module.rest_api.api_id
   resource_id   = aws_api_gateway_resource.proxy_resource.id
   http_method   = "ANY"
   # ... rest of configuration
 }

# Add individual Lambda functions for each operation using the same bundle
 resource "aws_lambda_function" "say_hello_handler" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-SayHello"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "sayHello.handler"  # Specific handler for this operation
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = var.env
   }

   tags = var.tags
 }

 resource "aws_lambda_function" "get_documentation_handler" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-GetDocumentation"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "getDocumentation.handler"  # Specific handler for this operation
   runtime         = "nodejs22.x"
   timeout         = 30
   source_code_hash = data.archive_file.lambda_zip.output_base64sha256

   tracing_config {
     mode = "Active"
   }

   environment {
     variables = var.env
   }

   tags = var.tags
 }

# Add specific resources and methods for each operation
 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"
 }

# Update deployment to depend on new integrations
~ 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,
     ]))
   }
  }

# Add Lambda permissions for each function
 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 模块自动使用路由器模式 - 这是默认且唯一支持的方法。生成的模块创建一个处理所有 API 操作的单个 Lambda 函数。

您可以简单地实例化默认模块以获得路由器模式：

# Default router pattern - single Lambda function for all operations
module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  asset_bucket_name = module.asset_bucket.bucket_name

  # Single Lambda function handles all operations automatically
  tags = local.common_tags
}

代码生成

CDK
Terraform

由于 FastAPI 中的操作是用 Python 定义的，而 CDK 基础设施是用 TypeScript 定义的，我们使用代码生成来向 CDK 构造提供元数据，以提供类型安全的集成接口。

一个 generate:<ApiName>-metadata 目标被添加到公共构造的 project.json 中以促进此代码生成，它会生成一个文件，例如 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 @<scope>/common-constructs:"generate:<ApiName>-metadata"

yarn nx watch --projects=<FastAPIProject> -- \
yarn nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

npx nx watch --projects=<FastAPIProject> -- \
npx nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

bunx nx watch --projects=<FastAPIProject> -- \
bunx nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

授予访问权限（仅限 IAM）

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

CDK
Terraform

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

# Create an IAM policy to allow invoking the API
resource "aws_iam_policy" "api_invoke_policy" {
  name        = "MyApiInvokePolicy"
  description = "Policy to allow invoking the FastAPI"

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

# Attach the policy to an IAM role (e.g., for authenticated users)
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
}

# Or attach to an existing role by name
resource "aws_iam_role_policy_attachment" "api_invoke_access_existing" {
  role       = "MyExistingRole"
  policy_arn = aws_iam_policy.api_invoke_policy.arn
}

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

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 serve my-api

yarn nx serve my-api

npx nx serve my-api

bunx nx serve my-api

这会启动一个本地 FastAPI 开发服务器，具有：

代码更改时自动重新加载
/docs 或 /redoc 处的交互式 API 文档
/openapi.json 处的 OpenAPI 架构

调用您的 FastAPI

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

连接

使用 connection 生成器将此项目与工作区中的其他项目集成。以下连接涉及此项目：

React 到 FastAPI从 React 网站调用 Python FastAPI

FastAPI 到 Python DynamoDB将 FastAPI 连接到 DynamoDB 表