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 Inherit The preferred IaC provider. By default this is inherited from your initial selection.
    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 はストリーミングレスポンスをサポートしていないため、FastAPI をサポートするプラットフォームにデプロイする必要があります。最も簡単なオプションは 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-x86',
                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',
            },
          },
        });
      }
    }

    実装

    Section titled “実装”

    ストリーミングをサポートするようにインフラストラクチャを更新したら、FastAPI でストリーミング API を実装できます。API は以下を行う必要があります:

    • StreamingResponse を返す
    • 各レスポンスチャンクの戻り値型を宣言する
    • API Connection を使用する場合は 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")

    消費

    Section titled “消費”

    ストリームレスポンスを消費するには、API Connection Generator を使用してストリームチャンクを反復処理するタイプセーフなメソッドを利用できます。

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

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

    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 “コード生成”

    FastAPI の操作は Python で定義され、CDK インフラストラクチャは TypeScript で記述されるため、統合のタイプセーフなインターフェースを提供するためコード生成を実施します。

    タイプセーフな統合を容易にするため、共通コンストラクトの 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 を呼び出すには api-connection ジェネレータを使用できます。