FastAPI
FastAPI は Python で API を構築するためのフレームワークです。
FastAPI ジェネレータは、AWS CDK または Terraform のインフラストラクチャ設定を備えた新しい FastAPI を作成します。生成されるバックエンドはサーバーレスデプロイ用に AWS Lambda を使用し、AWS API Gateway API を介して公開されます。AWS Lambda Powertools を設定し、ロギング、AWS X-Ray トレーシング、Cloudwatch メトリクスを含むオブザーバビリティを実現します。
FastAPI の生成
Section titled “FastAPI の生成”新しい FastAPI は2つの方法で生成できます:
- インストール Nx Console VSCode Plugin まだインストールしていない場合
- VSCodeでNxコンソールを開く
- クリック
Generate (UI)"Common Nx Commands"セクションで - 検索
@aws/nx-plugin - py#api - 必須パラメータを入力
- framework: fastapi
- クリック
Generate
pnpm nx g @aws/nx-plugin:py#api --framework=fastapiyarn nx g @aws/nx-plugin:py#api --framework=fastapinpx nx g @aws/nx-plugin:py#api --framework=fastapibunx nx g @aws/nx-plugin:py#api --framework=fastapi変更されるファイルを確認するためにドライランを実行することもできます
pnpm nx g @aws/nx-plugin:py#api --framework=fastapi --dry-runyarn nx g @aws/nx-plugin:py#api --framework=fastapi --dry-runnpx nx g @aws/nx-plugin:py#api --framework=fastapi --dry-runbunx nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run| パラメータ | 型 | デフォルト | 説明 |
|---|---|---|---|
| name 必須 | string | - | 生成するAPIプロジェクトの名前 |
| framework | fastapi | fastapi | 使用するAPIフレームワーク。 |
| integrationPattern | isolated | shared | isolated | API用にAPI Gateway統合を生成する方法。isolated(デフォルト)またはsharedから選択します。 |
| auth | iam | cognito | custom | iam | APIの認証に使用する方法。iam(デフォルト)、cognito、customから選択します。 |
| directory | string | packages | アプリケーションを保存するディレクトリ |
| subDirectory | string | - | プロジェクトが配置されるサブディレクトリ。デフォルトではプロジェクト名になります。 |
| iac | inherit | cdk | terraform | inherit | 優先するIaCプロバイダー。デフォルトでは、初期選択から継承されます。 |
| moduleName | string | - | Pythonモジュール名 |
| infra | rest-lambda | http-lambda | none | rest-lambda | このAPIをデプロイするために使用するインフラストラクチャのタイプ。 |
| preferInstallDependencies | boolean | true | ジェネレーター実行後に依存関係のインストールを優先するかどうか。複数のジェネレーターをバッチ処理する際にインストールを延期する場合はfalseに設定します(後続のジェネレーターがNxプロジェクトグラフを計算できるよう、必要に応じてインストールは実行されます)。最後に一度だけインストールします。 |
ジェネレータの出力
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 プロジェクトのビルドターゲットと設定
Directorypackages/common/terraform
Directorysrc
Directoryapp/ プロジェクト/ジェネレータ固有のインフラストラクチャ用Terraformモジュール
- …
Directorycore/
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コンストラクト用のユーティリティ
Directorypackages/common/terraform/src
Directoryapp
Directoryapis
Directory<project-name>
- <project-name>.tf APIをデプロイするためのモジュール
Directorycore
Directoryapi
Directoryhttp-api
- http-api.tf HTTP APIをデプロイするモジュール(HTTP APIをデプロイする選択をした場合)
Directoryrest-api
- rest-api.tf REST APIをデプロイするモジュール(REST APIをデプロイする選択をした場合)
アーキテクチャ
Section titled “アーキテクチャ”デプロイされたアプリケーションは以下のアーキテクチャを持ちます:
REST APIには、API Gatewayステージの前にAWS WAFv2 Web ACLが含まれており、AWSマネージドデフォルトルールセットが有効になっています。
HTTP APIは直接WAFをサポートしていません。WAF保護が必要な場合は、代わりにREST APIを選択するか、HTTP APIの前にCloudFrontディストリビューションを配置してください。
FastAPI の実装
Section titled “FastAPI の実装”メインの API 実装は main.py にあります。ここで API ルートとその実装を定義します。例:
from pydantic import BaseModelfrom .init import app, tracer
class Item(BaseModel): name: str
@app.get("/items/{item_id}")@tracer.capture_methoddef get_item(item_id: int) -> Item: return Item(name=...)
@app.post("/items")@tracer.capture_methoddef create_item(item: Item): return ...ジェネレータは以下の機能を自動的に設定します:
- オブザーバビリティのための AWS Lambda Powertools 統合
- エラーハンドリングミドルウェア
- リクエスト/レスポンス相関ID
- メトリクス収集
- Lambda Web Adapter と uvicorn による AWS Lambda デプロイ
- タイプセーフなストリーミング (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 コンテキスト情報
- コールドスタートインジケータ
トレーシング
Section titled “トレーシング”AWS X-Ray トレーシングが自動的に設定されます。トレースにカスタムサブセグメントを追加できます:
from .init import app, tracer
@app.get("/items/{item_id}")@tracer.capture_methoddef read_item(item_id: int): # 新しいサブセグメントを作成 with tracer.provider.in_subsegment("fetch-item-details"): # ロジックをここに記述 return {"item_id": item_id}リクエストごとに CloudWatch メトリクスが自動収集されます。カスタムメトリクスを追加できます:
from .init import app, metricsfrom 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}未処理の例外はミドルウェアで捕捉され:
- スタックトレース付きで例外をログ記録
- 失敗メトリクスを記録
- クライアントに安全な 500 レスポンスを返却
- 相関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 のどちらをデプロイしたかの両方に依存します。
IAM 認証の場合、API Gateway リクエストコンテキストから抽出した sub を使用して Cognito で呼び出し元を検索します。main.py と同じ場所に identity.py を作成します:
import jsonimport osfrom typing import Annotated
from boto3 import clientfrom fastapi import Depends, HTTPException, Requestfrom 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)]import jsonimport osfrom typing import Annotated
from boto3 import clientfrom fastapi import Depends, HTTPException, Requestfrom 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) amr = ( request_context.get("authorizer", {}) .get("iam", {}) .get("cognitoIdentity", {}) .get("amr", []) ) sign_in = next((s for s in amr if ":CognitoSignIn:" in s), None) sub = sign_in.split(":")[-1] if sign_in 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' の場合、API Gateway Cognito User Pools オーソライザーは呼び出し元が Authorization ヘッダーで提供する JWT を検証し、検証済みのクレームをリクエストコンテキストに配置します。
main.py と同じ場所に identity.py を作成します:
import jsonfrom typing import Annotated
from fastapi import Depends, HTTPException, Requestfrom 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)]HTTP API は JWT オーソライザーを使用し、検証済みのクレームを authorizer.jwt.claims に配置します:
import jsonfrom typing import Annotated
from fastapi import Depends, HTTPException, Requestfrom 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("jwt", {}).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, Identityfrom .init import app, tracer
@app.get("/me")@tracer.capture_methoddef me(identity: CurrentUser) -> Identity: return identityストリーミング
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 BaseModelfrom .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 クラスは:
- Pydantic モデルを JSON Lines 形式 (
application/jsonl) にシリアライズします itemSchemaを含む正しい OpenAPI スキーマを生成するopenapi_responseヘルパーを提供し、connectionジェネレータがタイプセーフなストリーミングクライアントメソッドを生成できるようにします
ストリームレスポンスを消費するには、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(), }); }}これにより以下が設定されます:
- FastAPI アプリケーションの各操作用 AWS Lambda 関数
- 関数トリガーとしての API Gateway HTTP/REST API
- IAM ロールと権限
- CloudWatch ロググループ
- X-Ray トレーシング設定
- CloudWatch メトリクスネームスペース
API デプロイ用の Terraform モジュールは common/terraform フォルダにあります。Terraform 設定で使用できます。
API モジュールは Lambda デプロイ zip を共有 S3 アセットバケットにステージングします — 詳細は Terraform インフラストラクチャガイドを参照してください。デプロイごとに core/asset-bucket モジュールを一度インスタンス化し、その bucket_name 出力を asset_bucket_name 入力を介してすべての API / Lambda モジュールに渡します:
module "asset_bucket" { source = "../../common/terraform/src/core/asset-bucket"}
module "my_api" { source = "../../common/terraform/src/app/apis/my-api"
asset_bucket_name = module.asset_bucket.bucket_name
# Lambda関数用環境変数 env = { ENVIRONMENT = var.environment LOG_LEVEL = "INFO" }
# 追加IAMポリシー(必要な場合) additional_iam_policy_statements = [ # APIに必要な追加権限をここに記述 ]
tags = local.common_tags}これにより以下が設定されます:
- 全FastAPIルートを処理するAWS Lambda関数
- 関数トリガーとしてのAPI Gateway HTTP/REST API
- IAMロールと権限
- CloudWatchロググループ
- X-Rayトレーシング設定
- CORS設定
Terraformモジュールは使用可能ないくつかの出力を提供します:
# APIエンドポイントにアクセスoutput "api_url" { value = module.my_api.stage_invoke_url}
# Lambda関数詳細にアクセスoutput "lambda_function_name" { value = module.my_api.lambda_function_name}
# 追加権限付与用IAMロールoutput "lambda_execution_role_arn" { value = module.my_api.lambda_execution_role_arn}モジュールに変数を渡すことでCORS設定をカスタマイズできます:
module "my_api" { source = "../../common/terraform/src/app/apis/my-api"
asset_bucket_name = module.asset_bucket.bucket_name
# カスタムCORS設定 cors_allow_origins = ["https://myapp.com", "https://staging.myapp.com"] cors_allow_methods = ["GET", "POST", "PUT", "DELETE"] cors_allow_headers = [ "authorization", "content-type", "x-custom-header" ]
tags = local.common_tags}REST APIの場合、生成されたコンストラクトはデフォルトでAWS WAFv2 Web ACLをAPI Gatewayステージに関連付けます。Web ACLはAWSマネージドデフォルトルールセット(AWSManagedRulesCommonRuleSetおよびAWSManagedRulesKnownBadInputsRuleSet)を使用し、OWASP Top 10を含む一般的なWebエクスプロイトに対する保護を提供します。WAFリクエストログはCloudWatch Logsグループに書き込まれます。
生成されたrest-apiコンストラクトを編集して、ルールを追加、削除、または調整できます(例えば、レートベースルールや追加のマネージドルールグループを追加する場合など)。
オプトアウトする場合(例えば、独自のWeb ACLをアタッチする場合)、enableWafをfalseに設定します:
const api = new MyApi(this, 'MyApi', { integrations: MyApi.defaultIntegrations(this).build(), enableWaf: false,});オプトアウトする場合(例えば、独自のWeb ACLをアタッチする場合)、enable_wafをfalseに設定します:
module "my_api" { source = "../../common/terraform/src/app/apis/my-api"
asset_bucket_name = module.asset_bucket.bucket_name enable_waf = false}アクセスログ
Section titled “アクセスログ”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(), },});アカウントロールはcore/api/api-gateway-accountモジュールによって管理されます。これは生成されたAPIモジュールによってインスタンス化されます。アカウントを冪等的に構成し、terraform destroyでリセットされることはありません。
生成されたAPIモジュール内のaws_api_gateway_stageリソースのaccess_log_settingsブロックを編集することで、アクセスログ形式をカスタマイズできます。
REST/HTTP API CDKコンストラクトは、各オペレーションの統合を定義するための型安全なインターフェースを提供するように設定されています。
デフォルト統合
Section titled “デフォルト統合”静的メソッドdefaultIntegrationsを使用して、各オペレーションに個別のAWS Lambda関数を定義するデフォルトパターンを利用できます:
new MyApi(this, 'MyApi', { integrations: MyApi.defaultIntegrations(this).build(),});Terraformモジュールはデフォルトで単一Lambda関数を使用するルーターパターンを採用します。追加設定は不要です:
module "my_api" { source = "../../common/terraform/src/app/apis/my-api"
asset_bucket_name = module.asset_bucket.bucket_name
# モジュールは自動的にすべてのAPIオペレーションを処理する # 単一のLambda関数を作成します tags = local.common_tags}統合へのアクセス
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');Terraformのルーターパターンでは単一のLambda関数のみが存在します。モジュール出力を通じてアクセスできます:
# 単一Lambda関数に追加権限を付与resource "aws_iam_role_policy" "additional_permissions" { name = "additional-api-permissions" role = module.my_api.lambda_execution_role_name
policy = jsonencode({ Version = "2012-10-17" Statement = [ { Effect = "Allow" Action = [ "s3:GetObject", "s3:PutObject" ] Resource = "arn:aws:s3:::my-bucket/*" } ] })}デフォルトオプションのカスタマイズ
Section titled “デフォルトオプションのカスタマイズ”デフォルト統合で作成されるLambda関数のオプションをカスタマイズするには、withDefaultOptionsメソッドを使用します。例えば、すべてのLambda関数をVPC内に配置する場合:
const vpc = new Vpc(this, 'Vpc', ...);
new MyApi(this, 'MyApi', { integrations: MyApi.defaultIntegrations(this) .withDefaultOptions({ vpc, }) .build(),});VPC設定などのオプションをカスタマイズするには、生成されたTerraformモジュールを編集します。例として、すべてのLambda関数にVPCサポートを追加する場合:
# VPC変数を追加variable "vpc_subnet_ids" { description = "Lambda関数用VPCサブネットIDのリスト" type = list(string) default = []}
variable "vpc_security_group_ids" { description = "Lambda関数用VPCセキュリティグループIDのリスト" type = list(string) default = []}
# Lambda関数リソースを更新resource "aws_lambda_function" "api_lambda" { # ... 既存の設定 ...
# VPC設定を追加 vpc_config { subnet_ids = var.vpc_subnet_ids security_group_ids = var.vpc_security_group_ids }}VPC設定付きでモジュールを使用:
module "my_api" { source = "../../common/terraform/src/app/apis/my-api"
asset_bucket_name = module.asset_bucket.bucket_name
# VPC設定 vpc_subnet_ids = [aws_subnet.private_a.id, aws_subnet.private_b.id] vpc_security_group_ids = [aws_security_group.lambda_sg.id]
tags = local.common_tags}オペレーションごとのオプションのカスタマイズ
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で置き換えたオペレーションにはオプションを指定できないことに注意してください。これらはもはやデフォルト統合を使用しないためです。
同じオペレーションがwithOperationOptionsとwithOverridesの両方でターゲットにされている場合、呼び出す順序に関係なく型エラーが発生します。
Terraformで特定のオペレーションのオプションをカスタマイズするには、生成されたTerraformモジュールを編集してオペレーションごとに個別のLambda関数を設定する必要があります(下記明示的統合セクション参照)。
統合のオーバーライド
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(),});必要に応じて、デフォルト統合を使用せずに各オペレーションに直接統合を指定できます。例えば、オペレーションごとに異なる統合タイプを使用する場合や、新しいオペレーション追加時に型エラーを受け取りたい場合に有用です:
new MyApi(this, 'MyApi', { integrations: { sayHello: { integration: new LambdaIntegration(...), }, getDocumentation: { integration: new HttpIntegration(...), }, },});Terraformで明示的なオペレーションごとの統合を行うには、生成されたアプリ固有モジュールを修正してデフォルトのプロキシ統合を置き換えます。
packages/common/terraform/src/app/apis/my-api/my-api.tfを編集:
- デフォルトのプロキシルートを削除(例:
resource "aws_apigatewayv2_route" "proxy_routes") - 単一Lambda関数を個別関数で置き換え
- オペレーションごとに特定の統合とルートを作成(同じZIPバンドルを再利用)
# デフォルトの単一Lambda関数を削除 resource "aws_lambda_function" "api_lambda" { filename = data.archive_file.lambda_zip.output_path function_name = "MyApiHandler" role = aws_iam_role.lambda_execution_role.arn handler = "index.handler" runtime = "nodejs22.x" timeout = 30 # ... その他の設定 }
# デフォルトのプロキシ統合を削除 resource "aws_apigatewayv2_integration" "lambda_integration" { api_id = module.http_api.api_id integration_type = "AWS_PROXY" integration_uri = aws_lambda_function.api_lambda.invoke_arn # ... その他の設定 }
# デフォルトのプロキシルートを削除 resource "aws_apigatewayv2_route" "proxy_routes" { for_each = toset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"]) api_id = module.http_api.api_id route_key = "${each.key} /{proxy+}" target = "integrations/${aws_apigatewayv2_integration.lambda_integration.id}" # ... その他の設定 }
# 同じバンドルを使用してオペレーションごとに個別Lambda関数を追加 resource "aws_lambda_function" "say_hello_handler" { filename = data.archive_file.lambda_zip.output_path function_name = "MyApi-SayHello" role = aws_iam_role.lambda_execution_role.arn handler = "sayHello.handler" # このオペレーション用の特定ハンドラ runtime = "nodejs22.x" timeout = 30 source_code_hash = data.archive_file.lambda_zip.output_base64sha256
tracing_config { mode = "Active" }
environment { variables = var.env }
tags = var.tags }
resource "aws_lambda_function" "get_documentation_handler" { filename = data.archive_file.lambda_zip.output_path function_name = "MyApi-GetDocumentation" role = aws_iam_role.lambda_execution_role.arn handler = "getDocumentation.handler" # このオペレーション用の特定ハンドラ runtime = "nodejs22.x" timeout = 30 source_code_hash = data.archive_file.lambda_zip.output_base64sha256
tracing_config { mode = "Active" }
environment { variables = var.env }
tags = var.tags }
# オペレーションごとに特定の統合を追加 resource "aws_apigatewayv2_integration" "say_hello_integration" { api_id = module.http_api.api_id integration_type = "AWS_PROXY" integration_uri = aws_lambda_function.say_hello_handler.invoke_arn payload_format_version = "2.0" timeout_milliseconds = 30000 }
resource "aws_apigatewayv2_integration" "get_documentation_integration" { api_id = module.http_api.api_id integration_type = "HTTP_PROXY" integration_uri = "https://example.com/documentation" integration_method = "GET" }
# オペレーションごとに特定のルートを追加 resource "aws_apigatewayv2_route" "say_hello_route" { api_id = module.http_api.api_id route_key = "POST /sayHello" target = "integrations/${aws_apigatewayv2_integration.say_hello_integration.id}" authorization_type = "AWS_IAM" }
resource "aws_apigatewayv2_route" "get_documentation_route" { api_id = module.http_api.api_id route_key = "GET /documentation" target = "integrations/${aws_apigatewayv2_integration.get_documentation_integration.id}" authorization_type = "NONE" }
# 各関数にLambda権限を追加 resource "aws_lambda_permission" "say_hello_permission" { statement_id = "AllowExecutionFromAPIGateway-SayHello" action = "lambda:InvokeFunction" function_name = aws_lambda_function.say_hello_handler.function_name principal = "apigateway.amazonaws.com" source_arn = "${module.http_api.api_execution_arn}/*/*" }
resource "aws_lambda_permission" "get_documentation_permission" { statement_id = "AllowExecutionFromAPIGateway-GetDocumentation" action = "lambda:InvokeFunction" function_name = aws_lambda_function.get_documentation_handler.function_name principal = "apigateway.amazonaws.com" source_arn = "${module.http_api.api_execution_arn}/*/*" }# デフォルトの単一Lambda関数を削除 resource "aws_lambda_function" "api_lambda" { filename = data.archive_file.lambda_zip.output_path function_name = "MyApiHandler" role = aws_iam_role.lambda_execution_role.arn handler = "index.handler" runtime = "nodejs22.x" timeout = 30 # ... その他の設定 }
# デフォルトのプロキシ統合を削除 resource "aws_apigatewayv2_integration" "lambda_integration" { api_id = module.http_api.api_id integration_type = "AWS_PROXY" integration_uri = aws_lambda_function.api_lambda.invoke_arn # ... その他の設定 }
# デフォルトのプロキシルートを削除 resource "aws_apigatewayv2_route" "proxy_routes" { for_each = toset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"]) api_id = module.http_api.api_id route_key = "${each.key} /{proxy+}" target = "integrations/${aws_apigatewayv2_integration.lambda_integration.id}" # ... その他の設定 }
# 同じバンドルを使用してオペレーションごとに個別Lambda関数を追加 resource "aws_lambda_function" "say_hello_handler" { filename = data.archive_file.lambda_zip.output_path function_name = "MyApi-SayHello" role = aws_iam_role.lambda_execution_role.arn handler = "sayHello.handler" # このオペレーション用の特定ハンドラ runtime = "nodejs22.x" timeout = 30 source_code_hash = data.archive_file.lambda_zip.output_base64sha256
tracing_config { mode = "Active" }
environment { variables = var.env }
tags = var.tags }
resource "aws_lambda_function" "get_documentation_handler" { filename = data.archive_file.lambda_zip.output_path function_name = "MyApi-GetDocumentation" role = aws_iam_role.lambda_execution_role.arn handler = "getDocumentation.handler" # このオペレーション用の特定ハンドラ runtime = "nodejs22.x" timeout = 30 source_code_hash = data.archive_file.lambda_zip.output_base64sha256
tracing_config { mode = "Active" }
environment { variables = var.env }
tags = var.tags }
# オペレーションごとに特定のリソースとメソッドを追加 resource "aws_api_gateway_resource" "say_hello_resource" { rest_api_id = module.rest_api.api_id parent_id = module.rest_api.api_root_resource_id path_part = "sayHello" }
resource "aws_api_gateway_method" "say_hello_method" { rest_api_id = module.rest_api.api_id resource_id = aws_api_gateway_resource.say_hello_resource.id http_method = "POST" authorization = "AWS_IAM" }
resource "aws_api_gateway_integration" "say_hello_integration" { rest_api_id = module.rest_api.api_id resource_id = aws_api_gateway_resource.say_hello_resource.id http_method = aws_api_gateway_method.say_hello_method.http_method
integration_http_method = "POST" type = "AWS_PROXY" uri = aws_lambda_function.say_hello_handler.invoke_arn }
resource "aws_api_gateway_resource" "get_documentation_resource" { rest_api_id = module.rest_api.api_id parent_id = module.rest_api.api_root_resource_id path_part = "documentation" }
resource "aws_api_gateway_method" "get_documentation_method" { rest_api_id = module.rest_api.api_id resource_id = aws_api_gateway_resource.get_documentation_resource.id http_method = "GET" authorization = "NONE" }
resource "aws_api_gateway_integration" "get_documentation_integration" { rest_api_id = module.rest_api.api_id resource_id = aws_api_gateway_resource.get_documentation_resource.id http_method = aws_api_gateway_method.get_documentation_method.http_method
integration_http_method = "GET" type = "HTTP" uri = "https://example.com/documentation" }
# デプロイメントの依存関係を更新~ resource "aws_api_gateway_deployment" "api_deployment" { rest_api_id = module.rest_api.api_id
depends_on = [ aws_api_gateway_integration.lambda_integration, aws_api_gateway_integration.say_hello_integration, aws_api_gateway_integration.get_documentation_integration, ]
lifecycle { create_before_destroy = true }
triggers = { redeployment = sha1(jsonencode([ aws_api_gateway_integration.say_hello_integration, aws_api_gateway_integration.get_documentation_integration, ])) } }
# 各関数にLambda権限を追加 resource "aws_lambda_permission" "say_hello_permission" { statement_id = "AllowExecutionFromAPIGateway-SayHello" action = "lambda:InvokeFunction" function_name = aws_lambda_function.say_hello_handler.function_name principal = "apigateway.amazonaws.com" source_arn = "${module.rest_api.api_execution_arn}/*/*" }
resource "aws_lambda_permission" "get_documentation_permission" { statement_id = "AllowExecutionFromAPIGateway-GetDocumentation" action = "lambda:InvokeFunction" function_name = aws_lambda_function.get_documentation_handler.function_name principal = "apigateway.amazonaws.com" source_arn = "${module.rest_api.api_execution_arn}/*/*" }統合パターン
Section titled “統合パターン”生成されたCDK APIコンストラクトは2つの統合パターンをサポートしています:
isolatedはオペレーションごとに1つのLambda関数を作成します。これが生成されたAPIのデフォルトです。sharedは単一のデフォルトルーターLambdaを作成し、特定の統合をオーバーライドしない限りすべてのオペレーションで再利用します。
isolatedはオペレーションごとにきめ細かい権限と設定を提供します。sharedはLambdaとAPI Gateway統合の増殖を抑えつつ、選択的なオーバーライドを可能にします。
例えば、patternを'shared'に設定すると、統合ごとに1つではなく単一の関数を作成します:
export class MyApi<...> extends ... {
public static defaultIntegrations = (scope: Construct) => { ... return IntegrationBuilder.rest({ pattern: 'shared', ... }); };}Terraformモジュールはデフォルトでルーターパターンを使用します - これはデフォルトかつ唯一のサポートされるアプローチです。生成されたモジュールは、すべてのAPIオペレーションを処理する単一のLambda関数を作成します。
デフォルトモジュールをインスタンス化するだけでルーターパターンを取得できます:
# デフォルトのルーターパターン - 全オペレーション用単一Lambda関数module "my_api" { source = "../../common/terraform/src/app/apis/my-api"
asset_bucket_name = module.asset_bucket.bucket_name
# 単一Lambda関数が全オペレーションを自動処理 tags = local.common_tags}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);# API呼び出しを許可するIAMポリシーを作成resource "aws_iam_policy" "api_invoke_policy" { name = "MyApiInvokePolicy" description = "FastAPI呼び出しを許可するポリシー"
policy = jsonencode({ Version = "2012-10-17" Statement = [ { Effect = "Allow" Action = "execute-api:Invoke" Resource = "${module.my_api.api_execution_arn}/*/*" } ] })}
# ポリシーをIAMロールにアタッチ(例: 認証済みユーザーロール)resource "aws_iam_role_policy_attachment" "api_invoke_access" { role = aws_iam_role.authenticated_user_role.name policy_arn = aws_iam_policy.api_invoke_policy.arn}
# 既存ロールにアタッチ(名前指定)resource "aws_iam_role_policy_attachment" "api_invoke_access_existing" { role = "MyExistingRole" policy_arn = aws_iam_policy.api_invoke_policy.arn}IAMポリシーで使用する主要な出力:
module.my_api.api_execution_arn- execute-api:Invoke 権限付与用module.my_api.api_arn- API Gateway ARNmodule.my_api.lambda_function_arn- Lambda関数ARN
ローカル開発
Section titled “ローカル開発”ジェネレータはローカル開発サーバーを設定します。以下で起動できます:
pnpm nx serve my-apiyarn nx serve my-apinpx nx serve my-apibunx nx serve my-apiこれにより以下を含むローカル FastAPI 開発サーバーが起動します:
- コード変更時の自動リロード
/docsまたは/redocのインタラクティブAPIドキュメント/openapi.jsonのOpenAPIスキーマ
FastAPI の呼び出し
Section titled “FastAPI の呼び出し”React ウェブサイトから API を呼び出すには connection ジェネレータを使用できます。
connectionジェネレータを使用して、このプロジェクトをワークスペース内の他のプロジェクトと統合できます。このプロジェクトに関連する接続は以下の通りです: