name: python-fastapi-patterns description: “FastAPI web框架模式。触发词:fastapi、api端点、依赖注入、pydantic模型、openapi、swagger、starlette、异步api、rest api、uvicorn。” compatibility: “FastAPI 0.100+, Pydantic v2, Python 3.10+。生产环境需要uvicorn。” allowed-tools: “Read Write Bash” depends-on: [python-typing-patterns, python-async-patterns] related-skills: [python-database-patterns, python-observability-patterns, python-pytest-patterns]
FastAPI 模式
使用 FastAPI 进行现代异步 API 开发。
基础应用
from fastapi import FastAPI
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用程序生命周期 - 启动和关闭。"""
# 启动
app.state.db = await create_db_pool()
yield
# 关闭
await app.state.db.close()
app = FastAPI(
title="我的API",
version="1.0.0",
lifespan=lifespan,
)
@app.get("/")
async def root():
return {"message": "Hello World"}
请求/响应模型
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
class UserCreate(BaseModel):
"""带验证的请求模型。"""
name: str = Field(..., min_length=1, max_length=100)
email: EmailStr
age: int = Field(..., ge=0, le=150)
class UserResponse(BaseModel):
"""响应模型。"""
id: int
name: str
email: EmailStr
created_at: datetime
model_config = {"from_attributes": True} # 启用 ORM 模式
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
db_user = await create_user_in_db(user)
return db_user
路径和查询参数
from fastapi import Query, Path
from typing import Annotated
@app.get("/users/{user_id}")
async def get_user(
user_id: Annotated[int, Path(..., ge=1, description="用户ID")],
):
return await fetch_user(user_id)
@app.get("/users")
async def list_users(
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 10,
search: str | None = None,
):
return await fetch_users(skip=skip, limit=limit, search=search)
依赖注入
from fastapi import Depends
from typing import Annotated
async def get_db():
"""数据库会话依赖。"""
async with async_session() as session:
yield session
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
"""认证并返回当前用户。"""
user = await authenticate_token(db, token)
if not user:
raise HTTPException(status_code=401, detail="无效令牌")
return user
# 用于复用的注解类型
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]
@app.get("/me")
async def get_me(user: CurrentUser):
return user
异常处理
from fastapi import HTTPException
from fastapi.responses import JSONResponse
# 内置 HTTP 异常
@app.get("/items/{item_id}")
async def get_item(item_id: int):
item = await fetch_item(item_id)
if not item:
raise HTTPException(status_code=404, detail="项目未找到")
return item
# 自定义异常处理器
class ItemNotFoundError(Exception):
def __init__(self, item_id: int):
self.item_id = item_id
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc: ItemNotFoundError):
return JSONResponse(
status_code=404,
content={"detail": f"项目 {exc.item_id} 未找到"},
)
路由组织
from fastapi import APIRouter
# users.py
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/")
async def list_users():
return []
@router.get("/{user_id}")
async def get_user(user_id: int):
return {"id": user_id}
# main.py
from app.routers import users, items
app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")
快速参考
| 功能 | 用法 |
|---|---|
| 路径参数 | @app.get("/items/{id}") |
| 查询参数 | def f(q: str = None) |
| 请求体 | def f(item: ItemCreate) |
| 依赖 | Depends(get_db) |
| 认证 | Depends(get_current_user) |
| 响应模型 | response_model=ItemResponse |
| 状态码 | status_code=201 |
额外资源
./references/dependency-injection.md- 高级依赖注入模式、作用域、缓存./references/middleware-patterns.md- 中间件链、CORS、错误处理./references/validation-serialization.md- Pydantic v2 模式、自定义验证器./references/background-tasks.md- 后台任务、异步工作器、调度
脚本
./scripts/scaffold-api.sh- 生成 API 端点样板代码
资源文件
./assets/fastapi-template.py- 生产就绪的 FastAPI 应用骨架
另请参阅
先决条件:
python-typing-patterns- Pydantic 模型和类型提示python-async-patterns- 异步端点模式
相关技能:
python-database-patterns- SQLAlchemy 集成python-observability-patterns- 日志、指标、追踪中间件python-pytest-patterns- 使用 TestClient 进行 API 测试