FastAPI 是一个现代、高性能的 Python Web 框架，基于 Starlette 和 Pydantic，特别适合构建 RESTful API。以下是一些 FastAPI 的高阶用法，帮助你充分发挥其潜力：
1. 依赖注入 (Dependency Injection)
FastAPI 提供了强大的依赖注入系统，可以在路径操作函数中复用代码、管理资源或执行初始化操作。
1
2
3
4
5
6
7
8
9
10
11
12
| from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 10):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return {"message": "Hello Items!", "params": commons}
|
- 优点: 可以将公共逻辑（如认证、数据库连接）提取到依赖中，保持代码 DRY（Don’t Repeat Yourself）。
- 进阶用法: 支持嵌套依赖、全局依赖（通过
app.dependencies）。
2. 自定义响应模型与动态校验
通过 Pydantic 模型，你可以定义复杂的输入输出校验逻辑，并动态调整返回内容。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| from pydantic import BaseModel, validator
from fastapi import FastAPI
app = FastAPI()
class Item(BaseModel):
name: str
price: float
@validator("price")
def price_must_be_positive(cls, value):
if value <= 0:
raise ValueError("Price must be positive")
return value
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
return item
|
- 高阶技巧: 使用
response_model_exclude_unset=True 排除未设置的字段，或者用 Union 类型支持多种返回模型。
3. 异步数据库操作
FastAPI 支持异步编程，结合 ORM（如 SQLAlchemy 的异步支持或 Tortoise ORM）可以显著提升性能。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
app = FastAPI()
engine = create_async_engine("sqlite+aiosqlite:///example.db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db():
async with AsyncSessionLocal() as session:
yield session
@app.get("/data/")
async def read_data(db: AsyncSession = Depends(get_db)):
result = await db.execute("SELECT * FROM some_table")
return result.fetchall()
|
- 注意: 确保数据库驱动支持异步（如
aiosqlite 或 asyncpg）。
4. 中间件 (Middleware)
FastAPI 支持自定义中间件，用于处理请求和响应的全局逻辑，例如日志、CORS 或认证。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
|
5. WebSocket 支持
FastAPI 支持 WebSocket，适合实时应用场景。
1
2
3
4
5
6
7
8
9
10
| from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message received: {data}")
|
- 高阶用法: 结合依赖注入管理 WebSocket 连接状态，或实现广播功能。
6. 背景任务 (Background Tasks)
对于耗时操作（如发送邮件、处理文件），可以使用背景任务异步执行。
1
2
3
4
5
6
7
8
9
10
11
12
| from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(f"{message}\n")
@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification queued"}
|
- 注意: 背景任务不会阻塞响应，但需要确保任务本身是线程安全的。
7. 自定义异常处理
FastAPI 允许你定义全局异常处理器，统一处理特定类型的错误。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return JSONResponse(
status_code=400,
content={"message": "Validation error", "details": exc.errors()},
)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 0:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
|
- 进阶: 可以为特定异常类（如数据库错误）定义独立的处理器。
8. OpenAPI 扩展
FastAPI 自动生成 OpenAPI 文档，你可以通过 tags、responses 等参数增强文档。
1
2
3
4
5
6
7
| from fastapi import FastAPI
app = FastAPI()
@app.get("/items/", tags=["items"], summary="Get all items", response_description="List of items")
async def read_items():
return [{"id": 1, "name": "Foo"}]
|
- 高阶技巧: 使用
openapi_extra 参数自定义 OpenAPI Schema，或集成外部工具（如 ReDoc）优化文档展示。