FastAPI
FastAPI는 Python으로 API를 구축하기 위한 프레임워크입니다.
FastAPI 생성기는 AWS CDK 또는 Terraform 인프라 설정과 함께 새로운 FastAPI를 생성합니다. 생성된 백엔드는 서버리스 배포를 위해 AWS Lambda를 사용하며, AWS API Gateway API를 통해 노출됩니다. 로깅, AWS X-Ray 추적 및 Cloudwatch Metrics를 포함한 관찰성을 위해 AWS Lambda Powertools를 설정합니다.
사용법
섹션 제목: “사용법”FastAPI 생성
섹션 제목: “FastAPI 생성”두 가지 방법으로 새로운 FastAPI를 생성할 수 있습니다:
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- 설치 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 부트스트랩 스크립트
디렉터리<module_name>
- __init__.py 모듈 초기화
- init.py FastAPI 앱을 설정하고 powertools 미들웨어를 구성
- main.py API 구현
디렉터리scripts
- generate_open_api.py FastAPI 앱에서 OpenAPI 스키마를 생성하는 스크립트
인프라
섹션 제목: “인프라”이 생성기는 선택한 iac를 기반으로 코드형 인프라를 제공하므로, 관련 CDK constructs 또는 Terraform 모듈을 포함하는 packages/common에 프로젝트를 생성합니다.
공통 코드형 인프라 프로젝트는 다음과 같이 구성됩니다:
디렉터리packages/common/constructs
디렉터리src
디렉터리app/ 프로젝트/생성기에 특정한 인프라를 위한 Constructs
- …
디렉터리core/
app의 constructs에서 재사용되는 일반 constructs- …
- index.ts
app에서 constructs를 내보내는 진입점
- project.json 프로젝트 빌드 타겟 및 구성
디렉터리packages/common/terraform
디렉터리src
디렉터리app/ 프로젝트/생성기에 특정한 인프라를 위한 Terraform 모듈
- …
디렉터리core/
app의 모듈에서 재사용되는 일반 모듈- …
- project.json 프로젝트 빌드 타겟 및 구성
API를 배포하기 위해 다음 파일들이 생성됩니다:
디렉터리packages/common/constructs/src
디렉터리app
디렉터리apis
- <project-name>.ts CDK construct for deploying your API
디렉터리core
디렉터리api
- 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
디렉터리packages/common/terraform/src
디렉터리app
디렉터리apis
디렉터리<project-name>
- <project-name>.tf Module for deploying your API
디렉터리core
디렉터리api
디렉터리http-api
- http-api.tf Module for deploying an HTTP API (if you selected to deploy an HTTP API)
디렉터리rest-api
- rest-api.tf Module for deploying a REST API (if you selected to deploy a REST API)
아키텍처
섹션 제목: “아키텍처”배포된 애플리케이션은 다음과 같은 아키텍처를 가집니다:
REST API는 API Gateway 스테이지 앞에 AWS 관리형 기본 규칙 세트가 활성화된 AWS WAFv2 Web ACL을 포함합니다.
HTTP API는 WAF를 직접 지원하지 않습니다 — WAF 보호가 필요한 경우 REST API를 선택하거나 HTTP API 앞에 CloudFront 배포를 배치하세요.
FastAPI 구현
섹션 제목: “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 통합
- 오류 처리 미들웨어
- 요청/응답 상관관계
- 메트릭 수집
- Lambda Web Adapter를 통한 uvicorn을 사용한 AWS Lambda 배포
- 타입 안전 스트리밍 (REST API만 해당)
AWS Lambda Powertools를 사용한 관찰성
섹션 제목: “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_methoddef 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, 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}기본 메트릭에는 다음이 포함됩니다:
- 요청 수
- 성공/실패 수
- 콜드 스타트 메트릭
- 라우트별 메트릭
오류 처리
섹션 제목: “오류 처리”생성기는 포괄적인 오류 처리를 포함합니다:
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에서 이를 읽어 호출자의 신원을 추출할 수 있습니다.
예를 들어, 호출 사용자에 대한 세부 정보를 반환하는 /me 엔드포인트를 추가해 보겠습니다. 라우트 간에 재사용할 수 있도록 추출을 FastAPI dependency로 구현하겠습니다. 요청 컨텍스트의 형태 — 따라서 신원을 추출하는 방법 — 는 선택한 auth 방법과 REST 또는 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: # 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 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: # 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를 생성하세요:
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: # 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는 검증된 클레임을 authorizer.jwt.claims 아래에 배치하는 JWT 권한 부여자를 사용합니다:
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: # 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)]그런 다음 호출자의 신원이 필요한 모든 라우트에 CurrentUser 종속성을 주입할 수 있습니다:
from .identity import CurrentUser, Identityfrom .init import app, tracer
@app.get("/me")@tracer.capture_methoddef me(identity: CurrentUser) -> Identity: return identity스트리밍
섹션 제목: “스트리밍”생성된 FastAPI는 REST API를 사용할 때 기본적으로 스트리밍 응답을 지원합니다. 인프라는 AWS Lambda Web Adapter를 사용하여 Lambda 내부에서 uvicorn을 통해 FastAPI를 실행하도록 구성되며, 모든 REST API 작업에 대해 API Gateway에서 ResponseTransferMode.STREAM을 사용하여 스트리밍이 비스트리밍 작업과 함께 작동할 수 있도록 합니다.
JsonStreamingResponse 사용
섹션 제목: “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 배포
섹션 제목: “FastAPI 배포”FastAPI 생성기는 선택한 iac에 따라 CDK 또는 Terraform 인프라 코드를 생성합니다. 이를 사용하여 FastAPI를 배포할 수 있습니다.
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를 배포하기 위한 Terraform 모듈은 common/terraform 폴더에 있습니다. Terraform 구성에서 이를 사용할 수 있습니다.
API 모듈은 공유 S3 자산 버킷에 Lambda 배포 zip을 스테이징합니다 — 자세한 내용은 Terraform 인프라 가이드를 참조하세요. 배포당 한 번 core/asset-bucket 모듈을 인스턴스화하고 asset_bucket_name 입력을 통해 모든 API / Lambda 모듈에 bucket_name 출력을 전달하세요:
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 모듈은 사용할 수 있는 여러 출력을 제공합니다:
# Access the API endpointoutput "api_url" { value = module.my_api.stage_invoke_url}
# Access Lambda function detailsoutput "lambda_function_name" { value = module.my_api.lambda_function_name}
# Access IAM role for granting additional permissionsoutput "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
섹션 제목: “WAF”REST API의 경우, 생성된 구성 요소는 기본적으로 AWS WAFv2 Web ACL을 API Gateway 스테이지와 연결합니다. Web ACL은 AWS 관리형 기본 규칙 세트(AWSManagedRulesCommonRuleSet 및 AWSManagedRulesKnownBadInputsRuleSet)를 사용하여 OWASP Top 10을 포함한 일반적인 웹 공격으로부터 보호합니다. 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}액세스 로깅
섹션 제목: “액세스 로깅”REST API의 경우, 생성된 인프라는 기본적으로 액세스 로깅을 활성화하여 요청당 하나의 구조화된 JSON 라인을 전용 CloudWatch Logs 그룹에 작성합니다. 로그 그룹은 고객 관리형 KMS 키로 암호화되며 1년 동안 보관됩니다.
API Gateway는 계정 수준의 CloudWatch Logs 역할을 사용하여 액세스 로그를 작성합니다. 이 역할은 AWS::ApiGateway::Account 설정에 구성되며, 이는 리전당 계정당 싱글톤입니다 — 리전의 모든 REST API에 대해 하나의 역할만 존재합니다. 독립적으로 배포된 여러 스택에서 이를 안전하게 관리하기 위해, 생성된 인프라는:
- 작동하는 역할이 이미 설정되어 있지 않은 경우에만 공유 CloudWatch Logs 역할을 생성하고 계정에 구성하므로, 배포가 다른 스택이 소유한 역할을 덮어쓰지 않습니다.
- 해체 시 계정 설정을 그대로 두므로, 하나의 스택을 삭제해도 리전의 다른 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(), },});계정 역할은 생성된 API 모듈에 의해 인스턴스화되는 core/api/api-gateway-account 모듈에 의해 관리됩니다. 이는 계정을 멱등적으로 구성하며 terraform destroy 시 재설정되지 않습니다.
생성된 API 모듈의 aws_api_gateway_stage 리소스에서 access_log_settings 블록을 편집하여 액세스 로그 형식을 사용자 지정할 수 있습니다.
REST/HTTP API 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}통합 액세스
섹션 제목: “통합 액세스”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 APIapi.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 함수가 하나만 있습니다. 모듈 출력을 통해 액세스할 수 있습니다:
# Grant additional permissions to the single Lambda functionresource "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/*" } ] })}기본 옵션 사용자 정의
섹션 제목: “기본 옵션 사용자 정의”각 기본 통합에 대해 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 지원을 추가하려면:
# Add VPC variablesvariable "vpc_subnet_ids" { description = "List of VPC subnet IDs for Lambda function" type = list(string) default = []}
variable "vpc_security_group_ids" { description = "List of VPC security group IDs for Lambda function" type = list(string) default = []}
# Update the Lambda function resourceresource "aws_lambda_function" "api_lambda" { # ... existing configuration ...
# Add VPC configuration 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 configuration 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}작업별 옵션 사용자 정의
섹션 제목: “작업별 옵션 사용자 정의”특정 작업에 대한 기본 통합을 생성하는 데 사용되는 옵션을 사용자 정의하려면(다른 작업에 영향을 주지 않고) withOperationOptions 메서드를 사용할 수 있습니다. 예를 들어, 하나의 작업에 대해서만 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 모두에서 동일한 작업을 대상으로 하는 경우 타입 오류가 발생합니다.
Terraform에서 특정 작업에 대한 옵션을 사용자 정의하려면 생성된 Terraform 모듈을 편집하여 작업별로 개별 Lambda 함수를 구성해야 합니다(아래 명시적 통합 섹션 참조).
통합 재정의
섹션 제목: “통합 재정의”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 mannerapi.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(...) // for REST, or HttpUserPoolAuthorizer for an HTTP API } }, }) .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 번들 재사용:
# Remove the default single Lambda function 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 # ... 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" { filename = data.archive_file.lambda_zip.output_path 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" { filename = data.archive_file.lambda_zip.output_path 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" { 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 # ... 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" { filename = data.archive_file.lambda_zip.output_path 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" { filename = data.archive_file.lambda_zip.output_path 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 API 구성은 두 가지 통합 패턴을 지원합니다:
isolated는 작업당 하나의 Lambda 함수를 생성합니다. 이것이 생성된 API의 기본값입니다.shared는 단일 기본 라우터 Lambda를 생성하고 특정 통합을 재정의하지 않는 한 모든 작업에 재사용합니다.
isolated는 작업별로 더 세밀한 권한과 구성을 제공합니다. shared는 선택적 재정의를 허용하면서 Lambda 및 API Gateway 통합 확산을 줄입니다.
예를 들어, pattern을 'shared'로 설정하면 통합당 하나가 아닌 단일 함수를 생성합니다:
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 operationsmodule "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}코드 생성
섹션 제목: “코드 생성”FastAPI의 작업은 Python으로 정의되고 CDK 인프라는 TypeScript로 정의되므로, 통합을 위한 타입 안전 인터페이스를 제공하기 위해 CDK 구성에 메타데이터를 제공하는 코드 생성을 계측합니다.
이 코드 생성을 용이하게 하기 위해 공통 구성 project.json에 generate:<ApiName>-metadata 타겟이 추가되며, packages/common/constructs/src/generated/my-api/metadata.gen.ts와 같은 파일을 생성합니다. 이는 빌드 시 생성되므로 버전 관리에서 무시됩니다.
액세스 권한 부여 (IAM만 해당)
섹션 제목: “액세스 권한 부여 (IAM만 해당)”IAM 인증을 사용하도록 선택한 경우, grantInvokeAccess 메서드를 사용하여 API에 대한 액세스 권한을 부여할 수 있습니다:
api.grantInvokeAccess(myIdentityPool.authenticatedRole);# Create an IAM policy to allow invoking the APIresource "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 nameresource "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 ARNmodule.my_api.lambda_function_arn- Lambda 함수 ARN
로컬 개발
섹션 제목: “로컬 개발”생성기는 다음 명령으로 실행할 수 있는 로컬 개발 서버를 구성합니다:
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 호출
섹션 제목: “FastAPI 호출”React 웹사이트에서 API를 호출하려면 connection 생성기를 사용할 수 있습니다.
connection 생성기를 사용하여 이 프로젝트를 워크스페이스의 다른 프로젝트와 통합하세요. 다음 연결은 이 프로젝트와 관련됩니다: