Backend API 需要一条从 request 到 database 的短路径，validation、documentation 与 infrastructure 来自同一份代码。这套 stack 是 FastAPI、Pydantic v2、SQLAlchemy 2、Alembic、PostgreSQL，以及跑在 AWS Lambda 或 ECS Fargate 上的 Terraform。Identity 是可移植的 OIDC access token，在 Lambda 上于 edge 验证，在 Fargate 上于 process 内验证。
Python 作为语言 —— duck typing、GIL、Depends 对比 Hono —— 见 以 TypeScript Developer 身份学 Python。SST 上的 TypeScript 兄弟篇是 用 Hono、Drizzle、Zod OpenAPI 与 SST 打造 Backend APIs。本站自己的 deploy loop 仍是 SST；这里的 Terraform 是独立的 architecture example，disclaimer 与 打造 Event-Driven 票务 Backend 相同。Tokens 见 OAuth 2.0 与 OIDC 解释。这篇笔记涵盖 typed FastAPI API，一路到 production 的 protect、monitor 与 recover。
1. Architecture
默认形态：一个 FastAPI Lambda 放在 API Gateway HTTP API 后面，使用内建 JWT authorizer。被拒绝的 tokens 永远不会 invoke function。同一个 app 在 Fargate 上用 uvicorn 跑。唯一需要知道 AWS 的文件，是 Mangum entry 与 Terraform。
| 选择 Lambda 的时机… | 选择 ECS Fargate 的时机… |
|---|---|
| 流量尖峰或闲置；handlers 短 | 稳定流量、长请求、streaming、workers |
| JWT authorizer 在 API 跑之前先拒绝 | 长驻 process 与真正的 connection pool |
| 薄 BFF 的最小 ops 表面 | 团队已经在交付 containers，并需要 warm capacity |
text
Client
→ Lambda: API Gateway → throttle → JWT authorizer → Mangum → FastAPI
→ Fargate: WAF → ALB → task → uvicorn → FastAPI Depends
→ Pydantic → SQLAlchemy → RDS Proxy → PostgreSQL
→ OpenAPI JSON → /docs
→ logs / metrics / alarms- 保持 entrypoint 精简。Routes、models 与 services 不要 import Mangum 或 boto3。
- 在 Lambda 上，identity 是 infrastructure 边界。在 Fargate 上，它在 request 打到 task 之后于 process 内执行。
- 两边都仍需要 edge controls。两边都经 RDS Proxy 访问 Postgres，避免一波 Lambda 在 instance 上为每次 invoke 各开一条 connection。
2. FastAPI
一个 app object。两套 process adapters。uv 掌管 lockfile。
pyproject.toml
[project]
name = "api"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115",
"pydantic[email]>=2.10",
"sqlalchemy[asyncio]>=2.0",
"asyncpg>=0.30",
"mangum>=0.19",
"pyjwt[crypto]>=2.10",
"uvicorn[standard]>=0.34",
"awslambdaric>=3.0",
]
[dependency-groups]
dev = ["pytest>=8.3", "httpx>=0.28", "alembic>=1.14"]app/main.py
from uuid import uuid4
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response
from app.errors import ApiError, error_response
from app.routers import users
from app.schemas import ErrorBody
app = FastAPI(title="Example API", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["https://app.example.com"],
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
allow_headers=["Content-Type", "Authorization", "X-Request-Id"],
allow_credentials=True,
)
@app.middleware("http")
async def request_id(request: Request, call_next):
rid = request.headers.get("x-request-id") or str(uuid4())
request.state.request_id = rid
response = await call_next(request)
response.headers["x-request-id"] = rid
return response
@app.get("/api/health", include_in_schema=False)
async def health() -> dict[str, str]:
return {"status": "ok"}
@app.options("/{path:path}")
async def options(path: str) -> Response:
return Response(status_code=204)
@app.exception_handler(RequestValidationError)
async def validation_error(request: Request, exc: RequestValidationError) -> JSONResponse:
body = ErrorBody(
code="VALIDATION_ERROR",
message="Request validation failed.",
request_id=getattr(request.state, "request_id", None),
issues=[
{"path": ".".join(str(p) for p in err["loc"]) or "request", "message": err["msg"]}
for err in exc.errors()
],
)
return JSONResponse(status_code=400, content=body.model_dump())
@app.exception_handler(ApiError)
async def api_error(request: Request, exc: ApiError) -> JSONResponse:
return error_response(request, exc.status_code, exc.code, str(exc.detail))
app.include_router(users.router)Lambda 是一个 module。Fargate 是 image 的 CMD。
app/lambda_handler.py
from mangum import Mangum
from app.main import app
handler = Mangum(app, lifespan="off")Dockerfile
FROM python:3.12-slim-bookworm
COPY --from=ghcr.io/astral-sh/uv:0.7.12 /uv /bin/uv
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
COPY app ./app
ENV PATH="/app/.venv/bin:$PATH"
EXPOSE 8000
# Fargate default. Lambda overrides command to the Mangum handler.
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]OPTIONS与/api/health保持 public。其余一切在正确边界走 auth。- Mangum 上的
lifespan="off"：SQLAlchemy engine 在 import 时创建，直觉与把 Drizzle client 建在 Hono handler 之外相同。Lambda 没有你可以指望的 graceful shutdown。 - AWS Lambda Web Adapter 也可以让 Lambda 上的 process 仍是
uvicorn。这篇笔记用 Mangum，让 adapter 成为明确的一行，如同 Hono 的handle(app)。
Failure: 从 app/routers/users.py import Mangum。这条 route 应当能在没有 AWS event 的情况下跑 pytest。
3. 用 Pydantic 定义 Contract
Contract 是一条 route 接受什么、返回什么，以及 errors 长什么样。一份 Pydantic model 同时驱动 validation、OpenAPI 与 response body。普通函数上的 type annotation 只是注释 —— 以 TypeScript Developer 身份学 Python。REST 形态见 System Design 里的 API Design。
app/schemas.py
from pydantic import BaseModel, EmailStr, Field
class UserRead(BaseModel):
id: str
email: EmailStr
name: str
class UserCreate(BaseModel):
email: EmailStr
name: str = Field(min_length=1, max_length=100)
class ErrorIssue(BaseModel):
path: str
message: str
class ErrorBody(BaseModel):
code: str
message: str
request_id: str | None = None
issues: list[ErrorIssue] | None = Noneapp/errors.py
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from app.schemas import ErrorBody
class ApiError(HTTPException):
def __init__(self, status_code: int, code: str, message: str) -> None:
super().__init__(status_code=status_code, detail=message)
self.code = code
def error_response(request: Request, status_code: int, code: str, message: str) -> JSONResponse:
body = ErrorBody(
code=code,
message=message,
request_id=getattr(request.state, "request_id", None),
)
return JSONResponse(status_code=status_code, content=body.model_dump())app/routers/users.py
from fastapi import APIRouter, Depends
from app.auth import Principal, require_user
from app.db import SessionDep
from app.schemas import ErrorBody, UserCreate, UserRead
from app.services import users as users_service
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/{user_id}", response_model=UserRead, responses={404: {"model": ErrorBody}})
async def get_user(
user_id: str,
session: SessionDep,
_: Principal = Depends(require_user),
) -> UserRead:
return await users_service.require(session, user_id)
@router.post("", response_model=UserRead, status_code=201, responses={409: {"model": ErrorBody}})
async def create_user(
body: UserCreate,
session: SessionDep,
_: Principal = Depends(require_user),
) -> UserRead:
return await users_service.create(session, body)response_model=UserRead是边界。Handler 返回 Pydantic model（或能 validate 成它的东西），不是 SQLAlchemy instance。- 共享的
ErrorBody让400/404/409保持一致。API Gateway JWT 失败用更小的 gateway 形状表示401/403—— 两边都要写进文档。 - 尽早挂上 request ID，并把它放进 logs 与 error responses。
- FastAPI 已经从这些 annotations 发出
/openapi.json。你不必手写第二份 contract。
Failure: 因为「FastAPI 会 serialize」就把 ORM row 返回。那是 detached instance、lazy load，或 500。
4. SQLAlchemy 与 PostgreSQL
PostgreSQL 是 durable authority。SQLAlchemy 2 让 schema 与 queries 保持 typed，同时不把 SQL 藏起来。Statement 模型见 SQL 核心概念。在 concurrency 下，优先用 database constraints（unique email、timestamptz），而不是 check-then-insert。
在 import 时把 engine 创建一次。在 Lambda 上，那是每个 warm environment，不是舰队级 pool。每种 runtime 都指向 RDS Proxy。Lambda 用 NullPool，避免 process 在被冻结的 environments 之间握住 client。Fargate 用小 pool，因为 task 是长驻 process。
app/db.py
import os
from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.pool import NullPool
RUNTIME = os.environ.get("RUNTIME", "fargate")
DATABASE_URL = os.environ["DATABASE_URL"]
CONNECT_ARGS = {"ssl": True, "statement_cache_size": 0}
engine = (
create_async_engine(DATABASE_URL, poolclass=NullPool, connect_args=CONNECT_ARGS)
if RUNTIME == "lambda"
else create_async_engine(
DATABASE_URL,
pool_size=5,
max_overflow=5,
pool_timeout=10,
connect_args=CONNECT_ARGS,
)
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_session() -> AsyncIterator[AsyncSession]:
async with SessionLocal() as session:
yield session
SessionDep = Annotated[AsyncSession, Depends(get_session)]app/models.py
from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.db import Base
class User(Base):
__tablename__ = "users"
id: Mapped[str] = mapped_column(String(36), primary_key=True)
email: Mapped[str] = mapped_column(Text, unique=True, nullable=False)
name: Mapped[str] = mapped_column(Text, nullable=False)app/services/users.py
from uuid import uuid4
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.errors import ApiError
from app.models import User
from app.schemas import UserCreate, UserRead
def to_read(row: User) -> UserRead:
return UserRead(id=row.id, email=row.email, name=row.name)
async def require(session: AsyncSession, user_id: str) -> UserRead:
row = await session.scalar(select(User).where(User.id == user_id))
if row is None:
raise ApiError(404, "USER_NOT_FOUND", "User not found.")
return to_read(row)
async def create(session: AsyncSession, body: UserCreate) -> UserRead:
row = User(id=str(uuid4()), email=body.email, name=body.name)
session.add(row)
try:
await session.commit()
except IntegrityError as exc:
await session.rollback()
raise ApiError(409, "EMAIL_TAKEN", "Email already exists.") from exc
await session.refresh(row)
return to_read(row)statement_cache_size=0关掉 asyncpg 的 prepared-statement cache。当 prepared statements 保持打开时，RDS Proxy 会把 client pin 到一条 session；pinning 加上 Lambda concurrency，就是耗尽 instance 的方式。- 每个 warm environment 的 reuse 不是 global pooling。限制 function 的 reserved concurrency。Proxy 是 multiplexor。
- Migrations 是 deploy step，绝不是 request path 的一部分。对 Proxy 或 migrator task 跑
uv run alembic upgrade head，审阅 SQL，在代码依赖新形状之前 apply。 - 无法原子出货的改动，用 expand-and-contract。
- Secrets 来自 Secrets Manager，不是 committed 的
.env。Lambda 与 task 在 boot 时读取DATABASE_URL。
Failure: 在 Lambda 上用默认大小为 5 的 QueuePool。一百个并发 cold starts 就是五百条 connections 穿过随后会 pin 的 Proxy。Lambda 的默认是 NullPool 加 Proxy。
5. OIDC JWT
这个 API 是 resource server。它不跑 login form。你不拥有的 issuer（Auth0、Okta、Cognito，或任何 OIDC provider）签发 access token。Issuer URL 与 audience 是 Terraform variables。Roles、grants，以及 ID token 与 access token 的区分，见 OAuth 2.0 与 OIDC 解释。
在 Lambda 上，HTTP API JWT authorizer 在 Mangum 跑 之前 检查 iss、aud、expiry 与 JWKS signature。缺失或无效的 Authorization 永远不会 invoke function。在 Fargate 上，同样的检查是 request 打到 task 之后的 FastAPI Depends。
app/auth.py
import os
from dataclasses import dataclass
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient
OIDC_ISSUER = os.environ["OIDC_ISSUER"]
OIDC_AUDIENCE = os.environ["OIDC_AUDIENCE"]
_jwks = PyJWKClient(f"{OIDC_ISSUER.rstrip('/')}/.well-known/jwks.json")
_bearer = HTTPBearer(auto_error=True)
@dataclass(frozen=True)
class Principal:
sub: str
raw: dict[str, object]
def decode_access_token(token: str) -> Principal:
signing_key = _jwks.get_signing_key_from_jwt(token)
claims = jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
audience=OIDC_AUDIENCE,
issuer=OIDC_ISSUER,
)
sub = claims.get("sub")
if not isinstance(sub, str) or not sub:
raise HTTPException(status_code=401, detail="Token is missing sub.")
return Principal(sub=sub, raw=claims)
def require_user(
creds: Annotated[HTTPAuthorizationCredentials, Depends(_bearer)],
) -> Principal:
try:
return decode_access_token(creds.credentials)
except jwt.PyJWTError as exc:
raise HTTPException(status_code=401, detail="Invalid access token.") from exc- Identity 是
Principal.sub。绝不是 client 的X-User-Id。 - Authentication 回答 谁。Authorization（roles、org membership、RLS）留在 app 里 —— 见 multi-tenant 后端笔记。
PyJWKClient会 cache keys。不要每次 request 都亲手 fetch JWKS。- 在 Lambda 上，你可以从
event.requestContext.authorizer.jwt.claims读取已经验证过的 claims，并跳过第二次 JWKS 调用。把require_user作为唯一路径更便于测试。Authorizer 仍然提供 reject-before-invoke 的性质。 - 让
/api/health与OPTIONS离开 authorizer。
Failure: 信任 client 的 X-User-Id header，或带 credentials 的 wildcard CORS origin。Failure: 把 ID token 当成 API credential。
6. OpenAPI
OpenAPI JSON 是可移植的 artifact（CI、typed clients、breaking-change checks）。FastAPI 的 /docs 是 presentation 层。
- FastAPI 从同一个
appobject 提供/openapi.json与/docs。你已经点名了response_model与responses。 - Bearer scheme 注册一次，让生成的文档与 JWT authorizer 一致：
python
from fastapi.openapi.utils import get_openapi
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
schema = get_openapi(title=app.title, version=app.version, routes=app.routes)
schema["components"]["securitySchemes"] = {
"BearerAuth": {"type": "http", "scheme": "bearer", "bearerFormat": "JWT"}
}
schema["security"] = [{"BearerAuth": []}]
app.openapi_schema = schema
return schema
app.openapi = custom_openapi- 在 production 对
/docs与/openapi.json做 stage-gate 或 authenticate。公开的 reference 就是免费侦察。 - 即使 FastAPI 在 Lambda 路径上不发出 gateway
401/403，也要写进文档 —— 那是 authorizer 发出的。
7. 用 Terraform 部署
本仓库用 SST 部署这个站点。下面的 Terraform 是独立的 architecture example，不是这里已经存在的基础设施。在真实的 platform repository 里 pin 并测试 provider version。
一份 image。两份 compute resources。共享 VPC、RDS、Proxy 与 OIDC variables。没有 MSK、没有 global Aurora、没有 purchase cells —— 那些属于 票务笔记。
infra/providers.tf
terraform {
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 6.33"
}
}
}
provider "aws" {
region = var.region
}
variable "oidc_issuer" {
type = string
}
variable "oidc_audience" {
type = string
}
variable "image_uri" {
type = string
}HTTP API、JWT authorizer、放在 VPC 里的 container Lambda，以便到达 Proxy。Health 与 preflight 保持 public。
infra/lambda.tf
resource "aws_apigatewayv2_api" "http" {
name = "api"
protocol_type = "HTTP"
cors_configuration {
allow_origins = ["https://app.example.com"]
allow_methods = ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
allow_headers = ["content-type", "authorization", "x-request-id"]
allow_credentials = true
}
}
resource "aws_apigatewayv2_authorizer" "jwt" {
api_id = aws_apigatewayv2_api.http.id
authorizer_type = "JWT"
identity_sources = ["$request.header.Authorization"]
name = "oidc"
jwt_configuration {
audience = [var.oidc_audience]
issuer = var.oidc_issuer
}
}
resource "aws_lambda_function" "api" {
function_name = "api"
role = aws_iam_role.lambda.arn
package_type = "Image"
image_uri = var.image_uri
timeout = 15
memory_size = 512
image_config {
command = ["app.lambda_handler.handler"]
entry_point = ["/app/.venv/bin/python", "-m", "awslambdaric"]
}
vpc_config {
subnet_ids = var.private_subnet_ids
security_group_ids = [aws_security_group.compute.id]
}
environment {
variables = {
RUNTIME = "lambda"
DATABASE_URL = jsondecode(data.aws_secretsmanager_secret_version.app_db.secret_string)["url"]
OIDC_ISSUER = var.oidc_issuer
OIDC_AUDIENCE = var.oidc_audience
}
}
}
resource "aws_apigatewayv2_stage" "default" {
api_id = aws_apigatewayv2_api.http.id
name = "$default"
auto_deploy = true
}
resource "aws_apigatewayv2_integration" "lambda" {
api_id = aws_apigatewayv2_api.http.id
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_function.api.invoke_arn
payload_format_version = "2.0"
}
resource "aws_apigatewayv2_route" "health" {
api_id = aws_apigatewayv2_api.http.id
route_key = "GET /api/health"
target = "integrations/${aws_apigatewayv2_integration.lambda.id}"
}
resource "aws_apigatewayv2_route" "options" {
api_id = aws_apigatewayv2_api.http.id
route_key = "OPTIONS /{proxy+}"
target = "integrations/${aws_apigatewayv2_integration.lambda.id}"
}
resource "aws_apigatewayv2_route" "default" {
api_id = aws_apigatewayv2_api.http.id
route_key = "$default"
authorization_type = "JWT"
authorizer_id = aws_apigatewayv2_authorizer.jwt.id
target = "integrations/${aws_apigatewayv2_integration.lambda.id}"
}
resource "aws_lambda_permission" "apigw" {
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.api.function_name
principal = "apigateway.amazonaws.com"
source_arn = "${aws_apigatewayv2_api.http.execution_arn}/*/*"
}
resource "aws_wafv2_web_acl_association" "http_api" {
resource_arn = aws_apigatewayv2_stage.default.arn
web_acl_arn = aws_wafv2_web_acl.api.arn
}Fargate 使用 同一份 image。Task command 是 uvicorn。TLS 与 WAF 坐在 ALB 上。
infra/ecs.tf
resource "aws_ecs_cluster" "api" {
name = "api"
}
resource "aws_ecs_task_definition" "api" {
family = "api"
requires_compatibilities = ["FARGATE"]
network_mode = "awsvpc"
cpu = "512"
memory = "1024"
execution_role_arn = aws_iam_role.exec.arn
task_role_arn = aws_iam_role.task.arn
container_definitions = jsonencode([{
name = "api"
image = var.image_uri
portMappings = [{ containerPort = 8000, protocol = "tcp" }]
environment = [
{ name = "RUNTIME", value = "fargate" },
{ name = "OIDC_ISSUER", value = var.oidc_issuer },
{ name = "OIDC_AUDIENCE", value = var.oidc_audience },
]
secrets = [
{ name = "DATABASE_URL", valueFrom = "${aws_secretsmanager_secret.app_db.arn}:url::" }
]
logConfiguration = {
logDriver = "awslogs"
options = {
awslogs-group = aws_cloudwatch_log_group.api.name
awslogs-region = var.region
awslogs-stream-prefix = "api"
}
}
healthCheck = {
command = ["CMD-SHELL", "python -c \"import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/api/health')\""]
interval = 30
timeout = 5
retries = 3
startPeriod = 15
}
}])
}
resource "aws_ecs_service" "api" {
name = "api"
cluster = aws_ecs_cluster.api.id
task_definition = aws_ecs_task_definition.api.arn
desired_count = 2
launch_type = "FARGATE"
network_configuration {
assign_public_ip = false
subnets = var.private_subnet_ids
security_groups = [aws_security_group.compute.id]
}
load_balancer {
target_group_arn = aws_lb_target_group.api.arn
container_name = "api"
container_port = 8000
}
deployment_minimum_healthy_percent = 100
deployment_maximum_percent = 200
}
resource "aws_lb" "api" {
name = "api"
load_balancer_type = "application"
subnets = var.public_subnet_ids
security_groups = [aws_security_group.alb.id]
}
resource "aws_lb_target_group" "api" {
name = "api"
port = 8000
protocol = "HTTP"
vpc_id = var.vpc_id
target_type = "ip"
health_check {
path = "/api/health"
}
}
resource "aws_lb_listener" "http" {
load_balancer_arn = aws_lb.api.arn
port = 80
protocol = "HTTP"
default_action {
type = "forward"
target_group_arn = aws_lb_target_group.api.arn
}
}
resource "aws_wafv2_web_acl" "api" {
name = "api"
scope = "REGIONAL"
default_action {
allow {}
}
rule {
name = "rate"
priority = 1
action {
block {}
}
statement {
rate_based_statement {
limit = 1000
aggregate_key_type = "IP"
}
}
visibility_config {
cloudwatch_metrics_enabled = true
metric_name = "api-rate"
sampled_requests_enabled = true
}
}
visibility_config {
cloudwatch_metrics_enabled = true
metric_name = "api"
sampled_requests_enabled = true
}
}
resource "aws_wafv2_web_acl_association" "alb" {
resource_arn = aws_lb.api.arn
web_acl_arn = aws_wafv2_web_acl.api.arn
}RDS 是 Multi-AZ。Proxy 坐在 private subnets。Function 与 task 都把 Proxy endpoint 当作 DATABASE_URL。
infra/rds.tf
resource "aws_db_instance" "postgres" {
engine = "postgres"
engine_version = "16"
instance_class = "db.t4g.medium"
allocated_storage = 20
db_name = "app"
username = "app"
manage_master_user_password = true
db_subnet_group_name = aws_db_subnet_group.this.name
vpc_security_group_ids = [aws_security_group.rds.id]
storage_encrypted = true
backup_retention_period = 7
deletion_protection = true
multi_az = true
performance_insights_enabled = true
}
resource "aws_db_proxy" "postgres" {
name = "app-proxy"
engine_family = "POSTGRESQL"
role_arn = aws_iam_role.proxy.arn
vpc_subnet_ids = var.private_subnet_ids
vpc_security_group_ids = [aws_security_group.proxy.id]
require_tls = true
auth {
auth_scheme = "SECRETS"
iam_auth = "DISABLED"
secret_arn = aws_db_instance.postgres.master_user_secret[0].secret_arn
}
}
resource "aws_db_proxy_default_target_group" "postgres" {
db_proxy_name = aws_db_proxy.postgres.name
connection_pool_config {
max_connections_percent = 90
max_idle_connections_percent = 50
connection_borrow_timeout = 30
}
}
resource "aws_db_proxy_target" "postgres" {
db_proxy_name = aws_db_proxy.postgres.name
target_group_name = aws_db_proxy_default_target_group.postgres.name
db_instance_identifier = aws_db_instance.postgres.identifier
}
# Compose postgresql+asyncpg://user:pass@PROXY:5432/app from the master
# secret and the proxy endpoint. Both runtimes read this key as DATABASE_URL.
resource "aws_secretsmanager_secret" "app_db" {
name = "api/database-url"
}
data "aws_secretsmanager_secret_version" "app_db" {
secret_id = aws_secretsmanager_secret.app_db.id
# First apply must write a version (composed URL) before the function can decode it.
}- Function 放进 VPC，只因为它必须到达 Proxy。Cold start 会随 ENI attach 变长；预先创建的 Hyperplane ENI 让第二次 invoke 更便宜。
- ALB snippet 监听 80，是为了让这一跳可见。Production 是 listener 上的 443 加上 ACM certificate。
- 明确的 public
OPTIONS与/api/health很重要：带 JWT authorizer 的$default会拒绝 browser preflight 与 liveness。 - 与 runtime 无关的 app hardening：严格 CORS、request IDs、structured logs（不含 tokens）、Pydantic validation、least-privilege 的 task 与 function roles。
- 用
aws_wafv2_web_acl_association把 WAFv2 挂到 HTTP API 的 stage ARN，方式与挂在 ALB 上相同。 - Alembic 在 CI 或一次性 task 里跑，早于 新 image 接收流量。
8. Monitor 与 Recover
Logs、metrics 与 traces 应当共享一个 request id。针对症状告警。SNS 是「告诉人」的 bus。
| Signal | Fargate | Lambda |
|---|---|---|
| Logs | Task awslogs → CloudWatch | Function log group |
| Metrics | ALB 5xx / latency, ECS CPU / memory | Errors, duration, throttles, concurrency |
| Alarms | Unhealthy hosts, p99, CPU | Error rate, throttles, timeout proximity |
infra/alarms.tf
resource "aws_sns_topic" "alerts" {
name = "api-alerts"
}
resource "aws_sns_topic_subscription" "email" {
topic_arn = aws_sns_topic.alerts.arn
protocol = "email"
endpoint = "oncall@example.com"
}
resource "aws_cloudwatch_metric_alarm" "lambda_errors" {
alarm_name = "api-lambda-errors"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = 2
metric_name = "Errors"
namespace = "AWS/Lambda"
period = 60
statistic = "Sum"
threshold = 5
treat_missing_data = "notBreaching"
alarm_actions = [aws_sns_topic.alerts.arn]
dimensions = {
FunctionName = aws_lambda_function.api.function_name
}
}
resource "aws_cloudwatch_metric_alarm" "alb_5xx" {
alarm_name = "api-alb-5xx"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = 2
metric_name = "HTTPCode_Target_5XX_Count"
namespace = "AWS/ApplicationELB"
period = 60
statistic = "Sum"
threshold = 10
treat_missing_data = "notBreaching"
alarm_actions = [aws_sns_topic.alerts.arn]
dimensions = {
LoadBalancer = aws_lb.api.arn_suffix
}
}- CloudTrail 与 GuardDuty 留在 account 层级 —— 确认它们是开的；不要在 Console 人手编辑 Terraform 管理的 resources。
- Durable state 几乎全是 Postgres。Compute 可以从 git + image tag 重新部署。
- 先写下 RPO 与 RTO，然后：自动化 backups，retention 对齐 RPO；产品重要时做 cross-region snapshot copy；versioned Alembic revisions，让 restore 后的 DB 能赶上。
- Restore drill：snapshot → 非 prod 的
DATABASE_URL→alembic upgrade head→/api/health+ 一条关键路径 → 记录真实 RTO。
Failure: 把 Multi-AZ 叫作「disaster recovery」。那是同一 region 内的 high availability，不是 cross-region DR。策略见 System Design 里的 Disaster Recovery。
9. 用 pytest 做单元测试
测试写在 Mangum 之下。TestClient 对同一个 app object 说 HTTP。它不需要 AWS event、container，或真实 issuer。Stub require_user，让缺失的 JWKS key 不成为 unit-test 的问题。Queries 与 migrations 仍需要可丢弃的 Postgres —— 那是另一套 suite。
命令是 uv run pytest。Fixtures 放在 conftest.py，让 route tests 与 service tests 共用同一套 overrides。
tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.auth import Principal, require_user
from app.main import app
def principal() -> Principal:
return Principal(sub="user_1", raw={"sub": "user_1"})
@pytest.fixture
def client() -> TestClient:
app.dependency_overrides[require_user] = principal
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
@pytest.fixture
def anon_client() -> TestClient:
with TestClient(app) as test_client:
yield test_clienttests/test_users.py
from fastapi.testclient import TestClient
def test_health(client: TestClient) -> None:
response = client.get("/api/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_create_user_rejects_bad_email(client: TestClient) -> None:
response = client.post("/users", json={"email": "not-an-email", "name": "Ada"})
assert response.status_code == 400
assert response.json()["code"] == "VALIDATION_ERROR"
def test_create_user_rejects_missing_bearer(anon_client: TestClient) -> None:
response = anon_client.post("/users", json={"email": "ada@example.com", "name": "Ada"})
assert response.status_code == 403HTTP tests 覆盖 contract。只做 ORM → Pydantic mapping 的 service function，可以不经过 TestClient 来断言：
tests/test_users_service.py
from app.models import User
from app.services.users import to_read
def test_to_read_maps_orm_fields() -> None:
row = User(id="u1", email="ada@example.com", name="Ada")
assert to_read(row).model_dump() == {
"id": "u1",
"email": "ada@example.com",
"name": "Ada",
}dependency_overrides是 seam。Overriderequire_user；当 write path 不能碰 Postgres 时，再 overrideget_session。除非你在测app/auth.py本身，否则不要 patchjwt.decode。HTTPBearer(auto_error=True)把缺失的Authorization变成 403，不是 401。断言 dependency 实际发出的 status。select/IntegrityError/ Alembic 留给可丢弃的 Postgres（或 Testcontainers）。那仍是 pytest。它不是 route 的 unit test。
Failure: 在每个 test 里拼一个 API Gateway event 再调用 handler。那测的是 Mangum，不是 API。
10. Lifecycle 与 trade-offs
Lambda 上的 POST /users：API Gateway → JWT authorizer → Mangum → request ID → Pydantic → require_user → SQLAlchemy → RDS Proxy → Postgres → 201。同一套 models 出现在 /openapi.json。
- 把 deployed tests 留给 authorizer 决策、CORS、secrets，以及 live URL 上的
/api/health。 - 可接受的 trade-offs：container-image 的 cold start（比 zip 大）、Proxy 多一跳、CPython 的 GIL 使 CPU-bound 工作不属于这条 request path（学 Python），以及你必须保护的 Terraform state —— 本站的 app stacks 仍在 SST。每一层都保持可替换。
Mental Model
text
Contract → Pydantic models + shared errors + generated OpenAPI
Auth → OIDC JWT authorizer (Lambda) or Depends (Fargate)
Data → SQLAlchemy 2 + NullPool/Proxy on Lambda + Alembic on deploy
Deploy → one image → HTTP API + Function or WAF + ALB + Service
Protect → WAF, rate limits, least-privilege secrets, docs gated
Monitor → requestId logs + runtime metrics + SNS alarms
Recover → RPO/RTO → backups → restore drills → new image + DNS
Tests → pytest + TestClient below Mangum; stub Depends; Postgres for SQL保持路径简短：validated request、trusted identity、typed query、durable constraint、generated docs、可审阅的 infra。依流量形态选择 Lambda 或 Fargate，把 WAF 与 identity 放在正确边界，针对症状告警，在 restore drill 符合你写下的 RPO/RTO 之前，不要说它 production-ready。
这条脊柱的 TypeScript 版本是 Hono 笔记。票务规模的 Fargate 是 event-driven 票务笔记。
Recap Q&A
阅读下一篇笔记
深入理解 NestJS