FastAPI

FastAPI はPythonでAPIを構築するためのフレームワークです。

FastAPIジェネレータは、AWS CDKインフラストラクチャが設定された新しいFastAPIプロジェクトを作成します。生成されるバックエンドはサーバーレスデプロイ用にAWS Lambdaを使用し、AWS API Gateway APIを介して公開されます。AWS Lambda Powertools を設定し、ロギング、AWS X-Rayトレーシング、CloudWatchメトリクスを含むオブザーバビリティ機能を実装します。

使用方法

FastAPIの生成

新しいFastAPIプロジェクトを2つの方法で生成できます:

VSCode

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

CLI

pnpm

pnpm nx g @aws/nx-plugin:py#fast-api

yarn

yarn nx g @aws/nx-plugin:py#fast-api

npm

npx nx g @aws/nx-plugin:py#fast-api

bun

bunx nx g @aws/nx-plugin:py#fast-api

変更されるファイルを確認するためにドライランを実行することもできます

pnpm

pnpm nx g @aws/nx-plugin:py#fast-api --dry-run

yarn

yarn nx g @aws/nx-plugin:py#fast-api --dry-run

npm

npx nx g @aws/nx-plugin:py#fast-api --dry-run

bun

bunx nx g @aws/nx-plugin:py#fast-api --dry-run

オプション

パラメータ	型	デフォルト	説明
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.
moduleName	string	-	Python module name

Note

APIのデプロイにはREST APIまたはHTTP APIのいずれかを選択できます。選択の参考にはAPI Gateway Developer Guideをご覧ください。

ジェネレータの出力

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

ジェネレータはまた、APIのデプロイに使用できるCDKコンストラクトをpackages/common/constructsディレクトリに作成します。

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を保持

Tip

api-connectionジェネレータを使用する場合、コード生成を改善するためAPI操作のレスポンスモデルを明示的に指定することを推奨。詳細はこちら。

ストリーミング

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

Note

エンドツーエンドの例についてはダンジョンアドベンチャーチュートリアルを参照

実装方法

インフラストラクチャをストリーミング対応に更新後、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")

消費方法

ストリームレスポンスを消費するには、API Connectionジェネレータを使用してタイプセーフなチャンク反復処理を実装できます。

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メトリクスネームスペース

Note

Cognito認証を選択した場合、APIコンストラクトにidentityプロパティを指定する必要があります:

import { MyApi, UserIdentity } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const identity = new UserIdentity(this, 'Identity');

    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
      identity,
    });
  }
}

UserIdentityコンストラクトはts#cloudscape-website-authジェネレータで生成可能

タイプセーフな統合

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で定義されるため、タイプセーフな統合インターフェースを提供するためメタデータ生成を実装しています。

共通コンストラクトのproject.jsonにgenerate:<ApiName>-metadataターゲットを追加し、packages/common/constructs/src/generated/my-api/metadata.gen.tsのようなファイルを生成します。ビルド時に生成されるため、バージョン管理から除外されます。

Note

APIを変更する際は、CDKコンストラクトが消費する型を最新化するためビルドを実行する必要があります。

pnpm

pnpm nx run-many --target build --all

yarn

yarn nx run-many --target build --all

npm

npx nx run-many --target build --all

bun

bunx nx run-many --target build --all

Tip

CDKインフラとFastAPIを同時に開発する場合、nx watchを使用してAPI変更の度に型を再生成できます:

pnpm

pnpm nx watch --projects=<FastAPIProject> -- \
pnpm nx run <InfraProject>:"generate:<ApiName>-metadata"

yarn

yarn nx watch --projects=<FastAPIProject> -- \
yarn nx run <InfraProject>:"generate:<ApiName>-metadata"

npm

npx nx watch --projects=<FastAPIProject> -- \
npx nx run <InfraProject>:"generate:<ApiName>-metadata"

bun

bunx nx watch --projects=<FastAPIProject> -- \
bunx nx run <InfraProject>:"generate:<ApiName>-metadata"

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

IAM認証を選択した場合、grantInvokeAccessメソッドでAPIへのアクセス権限を付与できます:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

ローカル開発

ジェネレータが設定するローカル開発サーバーを以下で起動:

pnpm

pnpm nx run my-api:serve

yarn

yarn nx run my-api:serve

npm

npx nx run my-api:serve

bun

bunx nx run my-api:serve

これにより以下を含むFastAPI開発サーバーが起動:

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

FastAPIの呼び出し

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