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ジェネレータを使用してタイプセーフな反復処理を実装できます。

    FastAPIのデプロイ

    Section titled “FastAPIのデプロイ”

    FastAPIジェネレータは選択したiacProviderに基づきCDKまたはTerraformのインフラストラクチャコードを作成します。これを使用してFastAPIをデプロイできます。

    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で記述されるため、タイプセーフなインテグレーションを提供するためメタデータを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ジェネレータを使用できます。