Skip to content
@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プロジェクトを2つの方法で生成できます:

  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.

    ジェネレータの出力

    ジェネレータは<directory>/<api-name>ディレクトリに以下のプロジェクト構造を作成します:

    • project.json プロジェクト設定とビルドターゲット
    • pyproject.toml Pythonプロジェクト設定と依存関係
    • Directory<module_name>
      • __init__.py モジュール初期化
      • init.py FastAPIアプリのセットアップとpowertoolsミドルウェアの設定
      • main.py API実装
    • Directoryscripts
      • 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. リクエスト/レスポンス相関ID
    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 must be positive")
        return {"item_id": item_id}

    未処理例外はミドルウェアで捕捉され:

    1. スタックトレース付きで例外を記録
    2. 失敗メトリクスを記録
    3. クライアントに安全な500レスポンスを返却
    4. 相関IDを保持

    ストリーミング

    FastAPIではStreamingResponseレスポンスタイプを使用してストリーミング応答を実装できます。

    インフラストラクチャ変更

    AWS API Gatewayはストリーミングレスポンスをサポートしていないため、AWS Lambda Function URLを使用する必要があります。生成されたcommon/constructs/src/app/apis/<name>-api.tsコンストラクトをFunction 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を追加

    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フォルダにデプロイ用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メソッドを使用して特定の操作の統合をオーバーライドできます。各オーバーライドは、HTTPまたはREST APIに適したCDK統合コンストラクトに型付けされたintegrationプロパティを指定する必要があります。例えば、外部サイトでホストされているドキュメントを指すように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(),
    });
    

    // 後ほど別ファイルで、定義したバケットプロパティに型安全にアクセス可能
    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の場合はCognitoUserPoolsAuthorizer、HTTP APIの場合はHttpUserPoolAuthorizer
            }
          },
        })
        .build(),
    });

    明示的な統合

    デフォルト統合を使用せず、各操作に直接統合を指定することも可能です。これは例えば、操作ごとに異なるタイプの統合を使用する必要がある場合や、新しい操作を追加した際に型エラーを受け取りたい場合に有用です：

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

    ルーターパターン

    すべてのAPIリクエストを処理する単一のLambda関数をデプロイする場合は、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 {
              // すべての統合で同じルーターLambdaハンドラーを参照
              integration: new LambdaIntegration(router),
            };
          },
        });
      };
    }

    必要に応じて、router関数をメソッド内で構築する代わりにdefaultIntegrationsのパラメータとして定義するなど、コードを変更することも可能です。

    コード生成

    FastAPIの操作はPythonで定義され、インフラはTypeScriptで定義されるため、型安全な統合インターフェースを提供するためメタデータをCDKコンストラクトに供給するコード生成を実装します。

    common/constructsの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開発サーバーが起動します:

    • コード変更時の自動リロード
    • /docsまたは/redocのインタラクティブAPIドキュメント
    • /openapi.jsonのOpenAPIスキーマ

    FastAPIの呼び出し

    ReactウェブサイトからAPIを呼び出すにはapi-connectionジェネレータを使用します。