Skip to content
@aws/nx-plugin
Blog

FastAPI

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

  1. インストール Nx Console VSCode Plugin まだインストールしていない場合
  2. VSCodeでNxコンソールを開く
  3. クリック Generate (UI) "Common Nx Commands"セクションで
  4. 検索 @aws/nx-plugin - py#fast-api
  5. 必須パラメータを入力
    • クリック Generate

    オプション

    Section titled “オプション”
    パラメータ 型 デフォルト 説明
    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.
    iacProvider string CDK The preferred IaC provider
    moduleName string - Python module name

    ジェネレータの出力

    Section titled “ジェネレータの出力”

    ジェネレータは <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スキーマを生成するスクリプト

    インフラストラクチャ

    Section titled “インフラストラクチャ”

    このジェネレータは選択した iacProvider に基づいてInfrastructure as Codeを生成するため、packages/common に関連するCDKコンストラクトまたはTerraformモジュールを含むプロジェクトを作成します。

    共通のInfrastructure as Codeプロジェクトは以下の構造を持ちます：

    • Directorypackages/common/constructs
      • Directorysrc
        • Directoryapp/ プロジェクト/ジェネレータ固有のインフラストラクチャ用コンストラクト
          • …
        • Directorycore/ app 内のコンストラクトで再利用される汎用コンストラクト
          • …
        • index.ts app からコンストラクトをエクスポートするエントリーポイント
      • project.json プロジェクトのビルドターゲットと設定

    APIをデプロイするために、以下のファイルが生成されます：

    • Directorypackages/common/constructs/src
      • Directoryapp
        • Directoryapis
          • <project-name>.ts APIをデプロイするためのCDKコンストラクト
      • Directorycore
        • Directoryapi
          • http-api.ts HTTP APIをデプロイするCDKコンストラクト（HTTP APIをデプロイする選択をした場合）
          • rest-api.ts REST APIをデプロイするCDKコンストラクト（REST APIをデプロイする選択をした場合）
          • utils.ts APIコンストラクト用のユーティリティ

    FastAPIの実装

    Section titled “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によるオブザーバビリティ

    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 “ストリーミング”

    FastAPIでは StreamingResponse レスポンスタイプでストリーミング応答を実装可能

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

    Section titled “インフラストラクチャ変更”

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

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

    実装

    Section titled “実装”

    ストリーミング対応インフラ更新後、FastAPIでストリーミング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")

    消費

    Section titled “消費”

    ストリームレスポンスを消費するには API接続ジェネレータ を使用

    FastAPIのデプロイ

    Section titled “FastAPIのデプロイ”

    iacProvider 選択に基づきCDK/Terraformインフラコードを生成。これを使用してデプロイ可能

    common/constructs フォルダのCDKコンストラクトを使用:

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

    設定内容:

    1. FastAPI操作毎のLambda関数
    2. API Gateway HTTP/REST APIトリガー
    3. IAMロールと権限
    4. CloudWatchロググループ
    5. X-Rayトレーシング設定
    6. CloudWatchメトリクスネームスペース

    インテグレーション

    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: [...],
    }));

    デフォルトオプションのカスタマイズ

    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 “統合のオーバーライド”

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

    // 後で別ファイルで、定義した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用、HTTP APIの場合はHttpUserPoolAuthorizer
            }
          },
        })
        .build(),
    });

    明示的統合

    Section titled “明示的統合”

    必要に応じて、デフォルト統合を使用せずに各オペレーションに直接統合を指定できます。例えば、オペレーションごとに異なる統合タイプを使用する場合や、新しいオペレーション追加時に型エラーを受け取りたい場合に有用です：

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

    ルーターパターン

    Section titled “ルーターパターン”

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

    必要に応じて、router関数をdefaultIntegrationsのパラメータとして定義するなど、他の方法でコードを修正することもできます。

    コード生成

    Section titled “コード生成”

    PythonのFastAPI操作とTypeScriptのCDKインフラ間で型安全な連携のため、メタデータ生成を実装

    common/constructs の project.json に generate:<ApiName>-metadata ターゲットを追加。ビルド時に packages/common/constructs/src/generated/my-api/metadata.gen.ts を生成（バージョン管理対象外）

    アクセス権付与（IAM認証時）

    Section titled “アクセス権付与（IAM認証時）”

    IAM 認証選択時、grantInvokeAccess メソッドでAPIアクセスを許可:

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    ローカル開発

    Section titled “ローカル開発”

    ジェネレータが設定する開発サーバを起動:

    Terminal window
    pnpm nx run my-api:serve

    起動時機能:

    • コード変更時の自動リロード
    • /docs または /redoc で対話型APIドキュメント
    • /openapi.json でOpenAPIスキーマ

    FastAPIの呼び出し

    Section titled “FastAPIの呼び出し”

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