Skip to content

FastAPI

Filter this guidePick generator option values to hide sections that don't apply.

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

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

新しい FastAPI は2つの方法で生成できます:

  1. インストール Nx Console VSCode Plugin まだインストールしていない場合
  2. VSCodeでNxコンソールを開く
  3. クリック Generate (UI) "Common Nx Commands"セクションで
  4. 検索 @aws/nx-plugin - py#api
  5. 必須パラメータを入力
    • framework: fastapi
  6. クリック Generate
パラメータデフォルト説明
name 必須string-生成するAPIプロジェクトの名前
framework fastapifastapi使用するAPIフレームワーク。
integrationPattern isolated | sharedisolatedAPI用にAPI Gateway統合を生成する方法。isolated(デフォルト)またはsharedから選択します。
auth iam | cognito | customiamAPIの認証に使用する方法。iam(デフォルト)、cognito、customから選択します。
directory stringpackagesアプリケーションを保存するディレクトリ
subDirectory string-プロジェクトが配置されるサブディレクトリ。デフォルトではプロジェクト名になります。
iac inherit | cdk | terraforminherit優先するIaCプロバイダー。デフォルトでは、初期選択から継承されます。
moduleName string-Pythonモジュール名
infra rest-lambda | http-lambda | nonerest-lambdaこのAPIをデプロイするために使用するインフラストラクチャのタイプ。
preferInstallDependencies booleantrueジェネレーター実行後に依存関係のインストールを優先するかどうか。複数のジェネレーターをバッチ処理する際にインストールを延期する場合はfalseに設定します(後続のジェネレーターがNxプロジェクトグラフを計算できるよう、必要に応じてインストールは実行されます)。最後に一度だけインストールします。

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

このジェネレータは選択した 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コンストラクト用のユーティリティ

デプロイされたアプリケーションは以下のアーキテクチャを持ちます:

ClientWAFAPI Gateway(REST API)LambdaCloudWatch(Logs, Metrics)X-Ray(Traces)

REST APIには、API Gatewayステージの前にAWS WAFv2 Web ACLが含まれており、AWSマネージドデフォルトルールセットが有効になっています。

メインの 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 によるオブザーバビリティ”

ジェネレータは 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を保持

呼び出しユーザーへのアクセス

Section titled “呼び出しユーザーへのアクセス”

API が認証で保護されている場合、ルートハンドラは誰が呼び出しているかを知る必要があることがよくあります。生成された FastAPI は Lambda Web Adapter を介して AWS Lambda 内で実行され、API Gateway リクエストコンテキストを JSON として x-amzn-request-context ヘッダーに転送します。FastAPI の Request からこれを読み取り、呼び出し元の ID を抽出できます。

例として、呼び出しユーザーの詳細を返す /me エンドポイントを追加しましょう。抽出処理を FastAPI dependency として実装し、ルート間で再利用できるようにします。リクエストコンテキストの形状 — したがって ID の抽出方法 — は、選択した auth メソッドと REST API または HTTP API のどちらをデプロイしたかの両方に依存します。

auth = iam

IAM 認証の場合、API Gateway リクエストコンテキストから抽出した sub を使用して Cognito で呼び出し元を検索します。main.py と同じ場所に identity.py を作成します:

import json
import os
from typing import Annotated
from boto3 import client
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel
cognito = client("cognito-idp")
class Identity(BaseModel):
sub: str
username: str
def get_identity(request: Request) -> Identity:
# Lambda Web Adapter は API Gateway リクエストコンテキストを JSON として転送します
request_context_header = request.headers.get("x-amzn-request-context")
if not request_context_header:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
request_context = json.loads(request_context_header)
provider = request_context.get("identity", {}).get("cognitoAuthenticationProvider")
sub = provider.split(":")[-1] if provider else None
if not sub:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
users = cognito.list_users(
# ユーザープール ID が Lambda 環境に設定されていることを前提とします
UserPoolId=os.environ["USER_POOL_ID"],
Limit=1,
Filter=f'sub="{sub}"',
).get("Users", [])
if len(users) != 1:
raise HTTPException(status_code=403, detail=f"No user found with subjectId {sub}")
return Identity(sub=sub, username=users[0]["Username"])
CurrentUser = Annotated[Identity, Depends(get_identity)]
auth = cognito

auth: 'cognito' の場合、API Gateway Cognito User Pools オーソライザーは呼び出し元が Authorization ヘッダーで提供する JWT を検証し、検証済みのクレームをリクエストコンテキストに配置します。

main.py と同じ場所に identity.py を作成します:

import json
from typing import Annotated
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel
class Identity(BaseModel):
sub: str
username: str
def get_identity(request: Request) -> Identity:
# Lambda Web Adapter は API Gateway リクエストコンテキストを JSON として転送します
request_context_header = request.headers.get("x-amzn-request-context")
if not request_context_header:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
request_context = json.loads(request_context_header)
claims = request_context.get("authorizer", {}).get("claims", {})
sub = claims.get("sub")
username = claims.get("username")
if not sub or not username:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
return Identity(sub=sub, username=username)
CurrentUser = Annotated[Identity, Depends(get_identity)]

その後、呼び出し元の ID が必要な任意のルートに CurrentUser 依存関係を注入できます:

from .identity import CurrentUser, Identity
from .init import app, tracer
@app.get("/me")
@tracer.capture_method
def me(identity: CurrentUser) -> Identity:
return identity
infra = rest-lambda

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

生成された 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 ジェネレータがタイプセーフなストリーミングクライアントメソッドを生成できるようにします

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

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 メトリクスネームスペース
auth = cognito
auth = custom
infra = rest-lambda

REST APIの場合、生成されたコンストラクトはデフォルトでAWS WAFv2 Web ACLをAPI Gatewayステージに関連付けます。Web ACLはAWSマネージドデフォルトルールセット(AWSManagedRulesCommonRuleSetおよびAWSManagedRulesKnownBadInputsRuleSet)を使用し、OWASP Top 10を含む一般的なWebエクスプロイトに対する保護を提供します。WAFリクエストログはCloudWatch Logsグループに書き込まれます。

生成されたrest-apiコンストラクトを編集して、ルールを追加、削除、または調整できます(例えば、レートベースルールや追加のマネージドルールグループを追加する場合など)。

オプトアウトする場合(例えば、独自のWeb ACLをアタッチする場合)、enableWaffalseに設定します:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
enableWaf: false,
});
infra = rest-lambda

REST APIの場合、生成されたインフラストラクチャはデフォルトでアクセスログを有効にし、リクエストごとに1行の構造化されたJSONを専用のCloudWatch Logsグループに書き込みます。ログループは顧客管理のKMSキーで暗号化され、1年間保持されます。

API Gatewayは、アカウントレベルのCloudWatch Logsロールを使用してアクセスログを書き込みます。このロールはAWS::ApiGateway::Account設定で構成されており、これはリージョンごと、アカウントごとのシングルトンです。つまり、リージョン内のすべてのREST APIに対して1つのロールしか存在しません。独立してデプロイされる複数のスタック間でこれを安全に管理するため、生成されたインフラストラクチャは次のようになっています:

  • 共有CloudWatch Logsロールを作成し、動作中のロールがまだ設定されていない場合にのみアカウントに構成するため、デプロイメントが他のスタックが所有するロールを上書きすることはありません。
  • 破棄時にアカウント設定をそのままにするため、1つのスタックを破棄してもリージョン内の他のREST APIのログ記録が無効になることはありません。

アカウントロールはApiGatewayAccountコンストラクトによって管理されます。これはApiGatewayAccount.ensure(scope)を介して解決されるスタックスコープのシングルトンです。各REST APIのステージはこれに依存しており、ロールはLambdaベースのカスタムリソースによって構成されます。

APIを構築する際にdeployOptionsを渡すことで、アクセスログ形式をカスタマイズできます:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
deployOptions: {
accessLogFormat: AccessLogFormat.clf(),
},
});

REST/HTTP API CDKコンストラクトは、各オペレーションの統合を定義するための型安全なインターフェースを提供するように設定されています。

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

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 “オペレーションごとのオプションのカスタマイズ”

_特定の_オペレーションのデフォルト統合を作成する際に使用するオプションをカスタマイズするには(他のオペレーションに影響を与えずに)、withOperationOptionsメソッドを使用します。例えば、1つのオペレーションだけLambda関数のタイムアウトを増やす場合:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this)
.withOperationOptions({
sayHello: {
timeout: Duration.seconds(60),
},
})
.build(),
});
// 選択されたオペレーションはデフォルト統合のままなので、それに応じて型付けされます:
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

指定したオプションはデフォルト統合オプション(およびwithDefaultOptionsで設定したオプション)とマージされます。withOverridesで置き換えたオペレーションにはオプションを指定できないことに注意してください。これらはもはやデフォルト統合を使用しないためです。

同じオペレーションがwithOperationOptionswithOverridesの両方でターゲットにされている場合、呼び出す順序に関係なく型エラーが発生します。

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

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

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

生成された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',
...
});
};
}

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

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

auth = iam

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

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

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

Terminal window
pnpm nx serve my-api

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

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

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

connectionジェネレータを使用して、このプロジェクトをワークスペース内の他のプロジェクトと統合できます。このプロジェクトに関連する接続は以下の通りです:

FastAPI
React to FastAPIReactウェブサイトからPython FastAPIを呼び出す
FastAPIAmazon DynamoDBPython
FastAPI to Python DynamoDBFastAPIをDynamoDBテーブルに接続する