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

FastAPI

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

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

使用方式

生成 FastAPI

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

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

    选项配置

    参数 类型 默认值 描述
    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.

    生成器输出

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

    • project.json 项目配置和构建目标
    • pyproject.toml Python 项目配置和依赖项
    • 文件夹<模块名称>
      • __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

    流式传输

    使用 FastAPI 可以通过 StreamingResponse 响应类型实现流式响应。

    基础设施变更

    由于 AWS API Gateway 不支持流式响应，需要将 FastAPI 部署到支持该功能的平台。最简单的选择是使用 AWS Lambda 函数 URL。为此可以替换生成的 common/constructs/src/app/apis/<名称>-api.ts 构造，改用支持函数 URL 的版本。

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

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

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

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

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

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

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

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

    实现

    更新基础设施支持流式传输后，可以在 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 指标命名空间

    类型安全集成

    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 的文件。由于这是构建时生成的，版本控制中会忽略该文件。

    授予访问权限（仅限 IAM）

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

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    本地开发

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

    Terminal window
    pnpm nx run my-api:serve

    该命令启动本地 FastAPI 开发服务器，包含：

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

    调用 FastAPI

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