FastAPI

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

FastAPIジェネレーターは、AWS CDKまたはTerraformインフラストラクチャのセットアップを含む新しいFastAPIを作成します。生成されたバックエンドは、サーバーレスデプロイメントにAWS Lambdaを使用し、AWS API Gateway APIを介して公開されます。観測性のためにAWS Lambda Powertoolsをセットアップし、ロギング、AWS X-Rayトレーシング、Cloudwatch Metricsを含みます。

使用方法

FastAPIの生成

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

CLI
VSCode

pnpm nx g @aws/nx-plugin:py#api --framework=fastapi

yarn nx g @aws/nx-plugin:py#api --framework=fastapi

npx nx g @aws/nx-plugin:py#api --framework=fastapi

bunx nx g @aws/nx-plugin:py#api --framework=fastapi

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

pnpm nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

yarn nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

npx nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

bunx nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run

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

オプション

パラメータ	型	デフォルト	説明
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プロジェクトグラフを計算できるよう、必要に応じてインストールは実行されます）。最後に一度だけインストールします。

ジェネレーターの出力

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

インフラストラクチャ

このジェネレーターは、選択した iac に基づいてインフラストラクチャをコードとして提供するため、関連する CDK コンストラクトまたは Terraform モジュールを含む packages/common にプロジェクトを作成します。

共通のインフラストラクチャコードプロジェクトは、次のように構成されています：

CDK
Terraform

Directorypackages/common/constructs
- Directorysrc
  - Directoryapp/ プロジェクト/ジェネレーター固有のインフラストラクチャ用のコンストラクト
    …
  - Directorycore/ app 内のコンストラクトによって再利用される汎用コンストラクト
    …
  - index.ts app からコンストラクトをエクスポートするエントリーポイント
- project.json プロジェクトのビルドターゲットと設定

Directorypackages/common/terraform
- Directorysrc
  - Directoryapp/ プロジェクト/ジェネレーター固有のインフラストラクチャ用の Terraform モジュール
    …
  - Directorycore/ app 内のモジュールによって再利用される汎用モジュール
    …
- project.json プロジェクトのビルドターゲットと設定

APIをデプロイするために、以下のファイルが生成されます：

CDK
Terraform

Directorypackages/common/constructs/src
- Directoryapp
  - Directoryapis
    <project-name>.ts CDK construct for deploying your API
- Directorycore
  - Directoryapi
    http-api.ts CDK construct for deploying an HTTP API (if you selected to deploy an HTTP API)
    rest-api.ts CDK construct for deploying a REST API (if you selected to deploy a REST API)
    utils.ts Utilities for the API constructs

アーキテクチャ

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

REST API
HTTP API

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

FastAPIの実装

メインの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 ...

ジェネレーターは以下の機能を自動的にセットアップします：

観測性のためのAWS Lambda Powertools統合
エラーハンドリングミドルウェア
リクエスト/レスポンスの相関
メトリクス収集
Lambda Web Adapterとuvicornを介したAWS Lambdaデプロイメント
型安全なストリーミング（REST APIのみ）

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
リクエストパス、マッチしたルート、メソッド

トレーシング

AWS X-Rayトレーシングは自動的に設定されます。トレースにカスタムサブセグメントを追加できます：

from .init import app, tracer

@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("fetch-item-details"):
        # Your logic here
        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}

デフォルトのメトリクスには以下が含まれます：

リクエスト数
成功/失敗数
ルートごとのメトリクス（<method> <path>のrouteディメンション経由）

エラーハンドリング

ジェネレーターには包括的なエラーハンドリングが含まれています：

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

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

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

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

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

REST API
HTTP API

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:
    # The Lambda Web Adapter forwards the API Gateway request context as 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(
        # Assumes user pool id is configured in lambda environment
        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 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:
    # The Lambda Web Adapter forwards the API Gateway request context as 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(
        # Assumes user pool id is configured in lambda environment
        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を作成します：

REST API
HTTP API

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:
    # The Lambda Web Adapter forwards the API Gateway request context as 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 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:
    # The Lambda Web Adapter forwards the API Gateway request context as 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, Identity
from .init import app, tracer

@app.get("/me")
@tracer.capture_method
def me(identity: CurrentUser) -> Identity:
    return identity

ストリーミング

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

`JsonStreamingResponse`の使用

生成された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クラスは：

PydanticモデルをJSON Lines形式（application/jsonl）にシリアライズ
itemSchemaを含む正しいOpenAPIスキーマを生成するopenapi_responseヘルパーを提供し、connectionジェネレーターが型安全なストリーミングクライアントメソッドを生成できるようにします

消費

レスポンスのストリームを消費するには、connectionジェネレーターを使用できます。これにより、ストリーミングされたチャンクを反復処理するための型安全なメソッドが提供されます。

FastAPIのデプロイ

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

CDK
Terraform

common/constructsフォルダーにAPIをデプロイするためのCDKコンストラクトがあります。これをCDKアプリケーションで使用できます：

import { MyApi } from '@my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // Add the api to your stack
    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 Gateway / API AWS Lambda統合（HTTP / REST API用）において、そのCloudFrontディストリビューションを唯一許可されたCORSオリジンとして設定できます。 APIのrestrictCorsToメソッドを、ウェブサイトコンストラクト、CloudFrontディストリビューション、オリジン文字列、またはそれらの組み合わせで呼び出すことができます。

import { MyApi, MyWebsite } from '@my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
    const website = new MyWebsite(this, 'MyWebsite');
    api.restrictCorsTo(website, 'http://localhost:4200');
  }
}

MyWebsiteコンストラクトは、ts#websiteジェネレーターを使用して生成できます。

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#website#authジェネレーターを使用して生成できます

APIをデプロイするためのTerraformモジュールはcommon/terraformフォルダーにあります。これをTerraform設定で使用できます。

APIモジュールは、共有S3アセットバケットにLambdaデプロイメントzipをステージングします — 詳細についてはTerraformインフラストラクチャガイドを参照してください。デプロイメントごとにcore/asset-bucketモジュールを1回インスタンス化し、その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

  # Environment variables for the Lambda function
  env = {
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }

  # Additional IAM policies if needed
  additional_iam_policy_statements = [
    # Add any additional permissions your API needs
  ]

  tags = local.common_tags
}

これにより以下がセットアップされます：

すべてのFastAPIルートを提供するAWS Lambda関数
関数トリガーとしてのAPI Gateway HTTP/REST API
IAMロールと権限
CloudWatchロググループ
X-Rayトレーシング設定
CORS設定

ソリューションにTerraform Webサイトモジュールが含まれている場合、そのCloudFrontドメイン名を使用してCORSを制限できます。 CloudFrontドメイン名<domain_name>が与えられた場合、APIをデプロイするためのTerraformモジュール内に以下を追加します

HTTP APIの場合、cors_allow_originsプロパティを["http://localhost:4200", "http://localhost:4300", "https://<domain name>"]に設定します。これにより、API GatewayのCORSがこのディストリビューションとローカルホストに制限されます。
REST APIの場合、ALLOWED_ORIGINS環境変数を"https://<domain_name>"に設定します。これにより、AWS Lambda統合において、CloudFrontディストリビューションが（ローカルホスト以外の）唯一許可されたCORSオリジンとして設定されます。この制限はプリフライトOPTIONSには適用されないことに注意してください - この問題への対処の優先順位付けに協力するには、このGitHub issueに+1してください。

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  asset_bucket_name = module.asset_bucket.bucket_name

  cors_allow_origins = ["http://localhost:4200", "http://localhost:4300", "https://<domain name>"] // Only required for HTTP API

  env = {
    ALLOWED_ORIGINS = "https://<domain name>" // Only required for REST API
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }
}

MyWebsiteコンストラクトは、ts#websiteジェネレーターを使用して生成できます

Cognito認証を使用することを選択した場合、Cognito設定を提供する必要があります：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  asset_bucket_name    = module.asset_bucket.bucket_name

  user_pool_id         = local.user_pool_id
  user_pool_client_ids = [local.client_id]

  env = {
    ENVIRONMENT = var.environment
    LOG_LEVEL   = "INFO"
  }

  tags = local.common_tags
}

適切なTerraformリソースまたはモジュールを使用してCognito User PoolとClientをセットアップできます。

Terraformモジュールは、使用できるいくつかの出力を提供します：

# Access the API endpoint
output "api_url" {
  value = module.my_api.stage_invoke_url
}

# Access Lambda function details
output "lambda_function_name" {
  value = module.my_api.lambda_function_name
}

# Access IAM role for granting additional permissions
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

  # Custom CORS configuration
  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
}

WAF

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

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

CDK
Terraform

オプトアウトするには（例えば、独自の 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
}

アクセスログ

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

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

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

CDK
Terraform

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

アクセスログフォーマットは、APIが拡張するRestApiコンストラクトによって設定されます。カスタマイズするには、生成されたpackages/common/constructs/src/app/apis/my-api.ts内でdeployOptionsをsuperに渡し、コンストラクトがすでに設定しているtracingEnabledを保持します：

super(scope, id, {
  apiName: 'MyApi',
  // ...
  deployOptions: {
    tracingEnabled: true,
    accessLogFormat: AccessLogFormat.clf(),
  },
  ...props,
});

AccessLogFormatはaws-cdk-lib/aws-apigatewayからインポートされます。設定しないものはすべて、コンストラクトのデフォルト（標準フィールドを持つJSON形式）を保持します。

アカウントロールはcore/api/api-gateway-accountモジュールによって管理されます。これは生成されたAPIモジュールによってインスタンス化されます。アカウントをべき等に構成し、terraform destroy時にリセットされることはありません。

生成されたAPIモジュール内のaws_api_gateway_stageリソースのaccess_log_settingsブロックを編集することで、アクセスログフォーマットをカスタマイズできます。

統合

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

CDK
Terraform

CDK コンストラクトは、以下に説明する完全な型安全インテグレーションサポートを提供します。

静的な 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

  # The module automatically creates a single Lambda function
  # that handles all API operations
  tags = local.common_tags
}

インテグレーションへのアクセス

CDK
Terraform

API コンストラクトの integrations プロパティを介して、基礎となる AWS Lambda 関数に型安全な方法でアクセスできます。例えば、API が sayHello という名前のオペレーションを定義していて、この関数にいくつかの権限を追加する必要がある場合、次のように実行できます：

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

// sayHello is typed to the operations defined in your 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 関数は 1 つだけです。モジュールの出力を介してアクセスできます：

# Grant additional permissions to the single Lambda function
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/*"
      }
    ]
  })
}

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

CDK
Terraform

各デフォルトインテグレーションの Lambda 関数を作成する際に使用されるオプションをカスタマイズしたい場合は、withDefaultOptions メソッドを使用できます。例えば、すべての Lambda 関数を Vpc 内に配置したい場合：

const vpc = new Vpc(this, 'Vpc', ...);

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withDefaultOptions({
      vpc,
    })
    .build(),
});

VPC 設定は生成されたモジュールによって既にサポートされています — enable_vpc を vpc_id および subnet_ids と共に設定すると、モジュールは Lambda 関数を VPC 内に、作成されたセキュリティグループの背後にデプロイします：

module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  asset_bucket_name = module.asset_bucket.bucket_name

  # VPC configuration
  enable_vpc = true
  vpc_id     = aws_vpc.main.id
  subnet_ids = [aws_subnet.private_a.id, aws_subnet.private_b.id]

  tags = local.common_tags
}

モジュールが公開していないオプションについては、生成された Terraform モジュール内の aws_lambda_function リソースを直接編集してください。

オペレーションごとのオプションのカスタマイズ

CDK
Terraform

_特定の_オペレーションのデフォルトインテグレーションを作成するために使用されるオプションをカスタマイズする（他のオペレーションに影響を与えずに）には、withOperationOptions メソッドを使用できます。例えば、1 つのオペレーションだけの Lambda 関数タイムアウトを増やしたい場合：

const api = new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOperationOptions({
      sayHello: {
        timeout: Duration.seconds(60),
      },
    })
    .build(),
});

// The selected operations remain default integrations, so they're still typed accordingly:
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

指定したオプションは、デフォルトインテグレーションオプション（および withDefaultOptions で設定されたオプション）とマージされます。withOverrides で置き換えたオペレーションにはオプションを指定できないことに注意してください。これらはデフォルトインテグレーションを使用しなくなるためです。

withOperationOptions と withOverrides の両方で同じオペレーションをターゲットにすると、呼び出す順序に関係なく型エラーが発生します。

インテグレーションのオーバーライド

CDK
Terraform

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

// Later, perhaps in another file, you can access the bucket property we defined
// in a type-safe manner
api.integrations.getFile.bucket.grantRead(...);

オーソライザーのオーバーライド

CDK
Terraform

インテグレーションで options を指定して、オーソライザーなどの特定のメソッドオプションをオーバーライドすることもできます。例えば、getDocumentation オペレーションに Cognito 認証を使用したい場合：

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getDocumentation: {
        integration: new HttpIntegration('https://example.com/documentation'),
        options: {
          authorizer: new CognitoUserPoolsAuthorizer(...) // for REST, or HttpUserPoolAuthorizer for an HTTP API
        }
      },
    })
    .build(),
});

明示的なインテグレーション

CDK
Terraform

必要に応じて、デフォルトインテグレーションを使用せず、各オペレーションに直接インテグレーションを提供することもできます。これは、例えば各オペレーションが異なるタイプのインテグレーションを使用する必要がある場合や、新しいオペレーションを追加する際に型エラーを受け取りたい場合に便利です：

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 バンドルを再利用：

HTTP API
REST API

# Remove the default single Lambda function
 resource "aws_lambda_function" "api_lambda" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApiHandler"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "index.handler"
   runtime         = "nodejs22.x"
   timeout         = 30
   # ... rest of configuration
 }

# Remove the default proxy integration
 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
   # ... rest of configuration
 }

# Remove the default proxy routes
 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}"
   # ... rest of configuration
 }

# Add individual Lambda functions for each operation using the same bundle
 resource "aws_lambda_function" "say_hello_handler" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-SayHello"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "sayHello.handler"  # Specific handler for this operation
   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" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-GetDocumentation"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "getDocumentation.handler"  # Specific handler for this operation
   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
 }

# Add specific integrations for each operation
 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"
 }

# Add specific routes for each operation
 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"
 }

# Add Lambda permissions for each function
 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}/*/*"
 }

# Remove the default single Lambda function
 resource "aws_lambda_function" "api_lambda" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApiHandler-${random_string.suffix.result}"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "index.handler"
   runtime         = "nodejs22.x"
   timeout         = 30
   # ... rest of configuration
 }

# Remove the default proxy integration
 resource "aws_api_gateway_integration" "lambda_integration" {
   rest_api_id = module.rest_api.api_id
   resource_id = aws_api_gateway_resource.proxy_resource.id
   http_method = aws_api_gateway_method.proxy_method.http_method
   integration_http_method = "POST"
   type                    = "AWS_PROXY"
   uri                     = aws_lambda_function.api_lambda.invoke_arn
   # ... rest of configuration
 }

# Remove the default catch-all proxy method
 resource "aws_api_gateway_method" "proxy_method" {
   rest_api_id   = module.rest_api.api_id
   resource_id   = aws_api_gateway_resource.proxy_resource.id
   http_method   = "ANY"
   # ... rest of configuration
 }

# Add individual Lambda functions for each operation using the same bundle
 resource "aws_lambda_function" "say_hello_handler" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-SayHello"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "sayHello.handler"  # Specific handler for this operation
   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" {
   s3_bucket         = aws_s3_object.lambda_zip.bucket
   s3_key            = aws_s3_object.lambda_zip.key
   s3_object_version = aws_s3_object.lambda_zip.version_id
   function_name    = "MyApi-GetDocumentation"
   role            = aws_iam_role.lambda_execution_role.arn
   handler         = "getDocumentation.handler"  # Specific handler for this operation
   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
 }

# Add specific resources and methods for each operation
 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"
 }

# Update deployment to depend on new integrations
~ 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,
     ]))
   }
  }

# Add Lambda permissions for each function
 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}/*/*"
 }

インテグレーションパターン

CDK
Terraform

生成された 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 関数を作成します。

デフォルトのモジュールをインスタンス化するだけで、ルーターパターンを取得できます：

# Default router pattern - single Lambda function for all operations
module "my_api" {
  source = "../../common/terraform/src/app/apis/my-api"

  asset_bucket_name = module.asset_bucket.bucket_name

  # Single Lambda function handles all operations automatically
  tags = local.common_tags
}

コード生成

CDK
Terraform

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

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

APIを変更するたびに、CDKコンストラクトが使用する型が最新であることを確認するためにビルドを実行する必要があります。

pnpm build

yarn build

npm run build

bun build

CDKインフラストラクチャとFastAPIの両方を積極的に作業している場合、nx watchを使用して、API変更を行うたびにこれらの型を再生成できます：

pnpm nx watch --projects=<FastAPIProject> -- \
pnpm nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

yarn nx watch --projects=<FastAPIProject> -- \
yarn nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

npx nx watch --projects=<FastAPIProject> -- \
npx nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

bunx nx watch --projects=<FastAPIProject> -- \
bunx nx run @<scope>/common-constructs:"generate:<ApiName>-metadata"

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

IAM認証を使用することを選択した場合、grantInvokeAccessメソッドを使用してAPIへのアクセスを付与できます：

CDK
Terraform

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

# Create an IAM policy to allow invoking the API
resource "aws_iam_policy" "api_invoke_policy" {
  name        = "MyApiInvokePolicy"
  description = "Policy to allow invoking the FastAPI"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = "execute-api:Invoke"
        Resource = "${module.my_api.api_execution_arn}/*/*"
      }
    ]
  })
}

# Attach the policy to an IAM role (e.g., for authenticated users)
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
}

# Or attach to an existing role by name
resource "aws_iam_role_policy_attachment" "api_invoke_access_existing" {
  role       = "MyExistingRole"
  policy_arn = aws_iam_policy.api_invoke_policy.arn
}

IAMポリシーに使用できるAPIモジュールの主要な出力は次のとおりです：

module.my_api.api_execution_arn - execute-api:Invoke権限を付与するため
module.my_api.api_arn - API Gateway ARN
module.my_api.lambda_function_arn - Lambda関数ARN

ローカル開発

ジェネレーターは、以下のコマンドで実行できるローカル開発サーバーを設定します：

pnpm nx serve my-api

yarn nx serve my-api

npx nx serve my-api

bunx nx serve my-api

これにより、以下を備えたローカルFastAPI開発サーバーが起動します：

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

FastAPIの呼び出し

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

接続

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

React to FastAPIReactウェブサイトからPython FastAPIを呼び出す

FastAPI to Python DynamoDBFastAPIをDynamoDBテーブルに接続