FastAPI高阶用法

FastAPI 是一个现代、高性能的 Python Web 框架，基于 Starlette 和 Pydantic，特别适合构建 RESTful API。以下是一些 FastAPI 的高阶用法，帮助你充分发挥其潜力：

1. 依赖注入 (Dependency Injection)

FastAPI 提供了强大的依赖注入系统，可以在路径操作函数中复用代码、管理资源或执行初始化操作。

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 模型，你可以定义复杂的输入输出校验逻辑，并动态调整返回内容。

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）可以显著提升性能。

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 或认证。

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

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

• 用途: 日志记录、请求修改、性能监控等。

---

5. WebSocket 支持

FastAPI 支持 WebSocket，适合实时应用场景。

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)

对于耗时操作（如发送邮件、处理文件），可以使用背景任务异步执行。

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 允许你定义全局异常处理器，统一处理特定类型的错误。

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 等参数增强文档。

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）优化文档展示。

---