跳转到内容
@aws/nx-plugin
Blog

FastAPI

Filter this guidePick generator option values to hide sections that don't apply.

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

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

使用方式

Section titled “使用方式”

生成 FastAPI

Section titled “生成 FastAPI”

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

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

选项配置

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

生成器输出

Section titled “生成器输出”

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

基础设施

Section titled “基础设施”

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

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

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

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

  • 文件夹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 构造的实用工具

Architecture

Section titled “Architecture”

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

ClientWAFAPI Gateway(REST API)LambdaCloudWatch(Logs, Metrics)X-Ray(Traces)

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

实现 FastAPI

Section titled “实现 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 ...

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

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

使用 AWS Lambda Powertools 实现可观测性

Section titled “使用 AWS Lambda Powertools 实现可观测性”

日志记录

Section titled “日志记录”

生成器使用 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 上下文信息
  • 冷启动指示器

追踪

Section titled “追踪”

自动配置 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}

指标

Section titled “指标”

自动收集每个请求的 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}

默认指标包括:

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

错误处理

Section titled “错误处理”

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

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}

未捕获的异常会被中间件捕获并:

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

访问调用用户

Section titled “访问调用用户”

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

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

auth = iam

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

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:
    # Lambda Web Adapter 将 API Gateway 请求上下文作为 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(
        # 假设用户池 ID 在 lambda 环境中配置
        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

使用 auth: 'cognito' 时,API Gateway Cognito 用户池授权器会验证调用者在 Authorization 标头中提供的 JWT,并将验证后的声明放置在请求上下文中。

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

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:
    # Lambda Web Adapter 将 API Gateway 请求上下文作为 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)]

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

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


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

流式传输

Section titled “流式传输”

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

使用 JsonStreamingResponse

Section titled “使用 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 类:

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

消费

Section titled “消费”

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

部署 FastAPI

Section titled “部署 FastAPI”

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

用于部署 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(),
    });
  }
}

此配置设置:

  1. 为 FastAPI 应用中的每个操作创建 AWS Lambda 函数
  2. 将 API Gateway HTTP/REST API 作为函数触发器
  3. IAM 角色和权限
  4. CloudWatch 日志组
  5. X-Ray 追踪配置
  6. CloudWatch 指标命名空间
auth = cognito
auth = custom
infra = rest-lambda

WAF

Section titled “WAF”

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

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

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

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

Access logging

Section titled “Access logging”

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

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

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

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

您可以通过在构造 API 时传递 deployOptions 来自定义访问日志格式：

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

集成

Section titled “集成”

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

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

默认集成

Section titled “默认集成”

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

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

访问集成

Section titled “访问集成”

您可以通过 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');

自定义默认选项

Section titled “自定义默认选项”

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

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


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

自定义每个操作的选项

Section titled “自定义每个操作的选项”

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

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


// 选定的操作仍然是默认集成，因此它们仍然具有相应的类型：
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

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

如果同一操作同时被 withOperationOptions 和 withOverrides 指定，无论调用顺序如何，都会遇到类型错误。

覆盖集成

Section titled “覆盖集成”

使用 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(...);

覆盖授权器

Section titled “覆盖授权器”

可在集成中提供 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(),
});

显式集成

Section titled “显式集成”

您可以选择不使用默认集成，直接为每个操作提供集成。这在需要不同集成类型或希望在添加新操作时收到类型错误提示时非常有用：

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

集成模式

Section titled “集成模式”

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

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

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

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

packages/common/constructs/src/app/apis/my-api.ts
export class MyApi<...> extends ... {


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

代码生成

Section titled “代码生成”

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

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

auth = iam

授予访问权限(仅限 IAM)

Section titled “授予访问权限(仅限 IAM)”

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

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

本地开发

Section titled “本地开发”

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

Terminal window
pnpm nx serve my-api

这将启动本地 FastAPI 开发服务器,包含:

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

调用 FastAPI

Section titled “调用 FastAPI”

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

连接

Section titled “连接”

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

FastAPI
React to FastAPI从 React 网站调用 Python FastAPI
FastAPIAmazon DynamoDBPython
FastAPI to Python DynamoDB将 FastAPI 连接到 DynamoDB 表