FastAPI

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

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

使用方式

生成 FastAPI

有两种方式可以生成新的 FastAPI 项目：

VSCode

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

CLI

pnpm

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

yarn

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

npm

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

bun

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

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

pnpm

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

yarn

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

npm

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

bun

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.
moduleName	string	-	Python module name

注意

您可以选择使用 REST API 或 HTTP API 来部署 API，具体选择可参考 API Gateway 开发人员指南 进行决策。

生成器输出

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

• project.json 项目配置和构建目标
• pyproject.toml Python 项目配置和依赖项
• 文件夹<module_name>

  • __init__.py 模块初始化
  • init.py 设置 FastAPI 应用并配置 powertools 中间件
  • main.py API 实现
• 文件夹scripts

  • generate_open_api.py 从 FastAPI 应用生成 OpenAPI 模式的脚本

生成器还会在 packages/common/constructs 目录下创建用于部署 API 的 CDK 构造。

实现 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 ...

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

1. 用于可观测性的 AWS Lambda Powertools 集成
2. 错误处理中间件
3. 请求/响应关联
4. 指标收集
5. 使用 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 必须为正数")
    return {"item_id": item_id}

未处理的异常会被中间件捕获并：

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

提示

如果使用 api-connection 生成器，建议为 API 操作指定响应模型以获得更好的代码生成效果。查看详情

流式传输

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

基础设施变更

由于 AWS API Gateway 不支持流式响应，您需要将 FastAPI 部署到支持此功能的平台。最简单的选择是使用 AWS Lambda 函数 URL。为此，可以替换生成的 common/constructs/src/app/apis/<name>-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',
        },
      },
    });
  }
}

注意

完整端到端示例请参考 地牢冒险教程

实现

更新基础设施以支持流式传输后，可以在 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 生成器在 common/constructs 文件夹中创建了用于部署 API 的 CDK 构造。您可以在 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 指标命名空间

注意

如果选择使用 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#cloudscape-website-auth 生成器 生成

类型安全集成

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

默认集成

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

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

访问集成

您可以通过 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: [...],
}));

自定义默认选项

如果要调整为每个默认集成创建 Lambda 函数时使用的选项，可以使用 withDefaultOptions 方法。例如，若希望所有 Lambda 函数都部署在 VPC 中：

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

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

覆盖集成

您还可以通过 withOverrides 方法覆盖特定操作的集成。每个覆盖必须指定类型正确的 integration 属性（对应 HTTP 或 REST API 的 CDK 集成构造）。该方法同样具有类型安全性。例如，若要将 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(...);

覆盖授权器

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

显式集成

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

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

路由模式

如果希望部署单个 Lambda 函数处理所有 API 请求，可以修改 API 的 defaultIntegrations 方法创建共享函数：

packages/common/constructs/src/app/apis/my-api.ts

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

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

您也可以采用其他自定义方式，例如将 router 函数作为 defaultIntegrations 的参数而非在方法内构造。

代码生成

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

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

注意

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

pnpm

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

yarn

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

npm

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

bun

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

提示

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

pnpm

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

yarn

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

npm

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

bun

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

授予访问权限（仅限 IAM）

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

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

本地开发

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

pnpm

pnpm nx run my-api:serve

yarn

yarn nx run my-api:serve

npm

npx nx run my-api:serve

bun

bunx nx run my-api:serve

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

• 代码变更自动重载
• 交互式 API 文档（位于 /docs 或 /redoc）
• OpenAPI 模式（位于 /openapi.json）

调用 FastAPI

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