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 - 生成するAPIプロジェクトの名前
    computeType string ServerlessApiGatewayRestApi このAPIをデプロイするために使用するコンピュートのタイプ。ServerlessApiGatewayRestApi（デフォルト）またはServerlessApiGatewayHttpApiから選択します。
    integrationPattern string isolated API用にAPI Gateway統合を生成する方法。isolated（デフォルト）またはsharedから選択します。
    auth string IAM APIで認証するために使用する方法。IAM（デフォルト）、CognitoまたはNoneから選択します。
    directory string packages アプリケーションを保存するディレクトリ
    subDirectory string - プロジェクトが配置されるサブディレクトリ。デフォルトではプロジェクト名になります。
    iacProvider string Inherit 優先するIaCプロバイダー。デフォルトでは、初期選択から継承されます。
    moduleName string - Pythonモジュール名

    ジェネレータの出力

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

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

    • project.json プロジェクト設定とビルドターゲット
    • pyproject.toml Python プロジェクト設定と依存関係
    • run.sh uvicorn 経由で FastAPI アプリを起動する Lambda Web Adapter ブートストラップスクリプト
    • 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 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. リクエスト/レスポンス相関ID
    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 “ストリーミング”

    生成された FastAPI は、REST API を使用する場合、標準でストリーミングレスポンスをサポートします。インフラストラクチャは AWS Lambda Web Adapter を使用して Lambda 内で uvicorn 経由で FastAPI を実行するように設定されており、すべての REST API 操作に対して API Gateway で ResponseTransferMode.STREAM が設定されているため、ストリーミング操作と非ストリーミング操作を併用できます。

    JsonStreamingResponse の使用

    Section titled “JsonStreamingResponse の使用”

    生成された init.py は、適切な OpenAPI スキーマ生成を伴うタイプセーフなストリーミングを提供する JsonStreamingResponse クラスをエクスポートします。これにより、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. itemSchema を含む正しい OpenAPI スキーマを生成する openapi_response ヘルパーを提供し、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 メトリクスネームスペース

    統合

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

    withOverridesメソッドを使用して特定のオペレーションの統合をオーバーライドできます。各オーバーライドは、HTTPまたはREST APIに適したCDK統合コンストラクトに型付けされたintegrationプロパティを指定する必要があります。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用、HTTP APIの場合はHttpUserPoolAuthorizer
            }
          },
        })
        .build(),
    });

    明示的統合

    Section titled “明示的統合”

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

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

    統合パターン

    Section titled “統合パターン”

    生成されたCDK APIコンストラクトは2つの統合パターンをサポートしています：

    • isolatedはオペレーションごとに1つのLambda関数を作成します。これが生成されたAPIのデフォルトです。
    • sharedは単一のデフォルトルーターLambdaを作成し、特定の統合をオーバーライドしない限りすべてのオペレーションで再利用します。

    isolatedはオペレーションごとにきめ細かい権限と設定を提供します。sharedはLambdaとAPI Gateway統合の増殖を抑えつつ、選択的なオーバーライドを可能にします。

    例えば、patternを'shared'に設定すると、統合ごとに1つではなく単一の関数を作成します：

    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 のようなファイルを生成します。ビルド時に生成されるためバージョン管理対象外です。

    アクセス権限付与（IAMのみ）

    Section titled “アクセス権限付与（IAMのみ）”

    IAM 認証を選択した場合、grantInvokeAccess メソッドを使用して API へのアクセスを許可できます:

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    ローカル開発

    Section titled “ローカル開発”

    ジェネレータはローカル開発サーバーを設定します。以下で起動できます:

    Terminal window
    pnpm nx run my-api:serve

    これにより以下を含むローカル FastAPI 開発サーバーが起動します:

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

    FastAPI の呼び出し

    Section titled “FastAPI の呼び出し”

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