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, FastAPIapp = 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, validatorfrom fastapi import FastAPIapp = 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 FastAPIfrom sqlalchemy.ext.asyncio import AsyncSession, create_async_enginefrom sqlalchemy.orm import sessionmakerapp = 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, Requestfrom fastapi.middleware.cors import CORSMiddlewareimport timeapp = 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, WebSocketapp = 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, FastAPIapp = 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, HTTPExceptionfrom fastapi.exceptions import RequestValidationErrorfrom fastapi.responses import JSONResponseapp = 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 FastAPIapp = 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）优化文档展示。