复杂场景处理

多级认证

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    if not is_valid_token(token):
        raise HTTPException(status_code=401, detail="Invalid token")
    return decode_token(token)

async def get_current_user(token_data: dict = Depends(verify_token), db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == token_data["user_id"]).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

async def require_admin(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, admin: User = Depends(require_admin)):
    # 只有管理员能执行
    return {"message": f"User {user_id} deleted"}

条件依赖

def get_query_or_body(
    q: str = None,
    data: dict = Body(None)
):
    if q:
        return {"source": "query", "value": q}
    elif data:
        return {"source": "body", "value": data}
    else:
        raise HTTPException(status_code=400, detail="Need query or body")

@app.post("/process")
async def process(input_data: dict = Depends(get_query_or_body)):
    return input_data

分页依赖

from typing import Optional

class Pagination:
    def __init__(
        self,
        page: int = 1,
        page_size: int = 10,
        max_page_size: int = 100
    ):
        if page < 1:
            raise HTTPException(status_code=400, detail="Page must be >= 1")
        if page_size < 1 or page_size > max_page_size:
            raise HTTPException(
                status_code=400,
                detail=f"Page size must be between 1 and {max_page_size}"
            )
        
        self.page = page
        self.page_size = page_size
        self.skip = (page - 1) * page_size
        self.limit = page_size

@app.get("/items/")
async def list_items(
    pagination: Pagination = Depends(),
    db: Session = Depends(get_db)
):
    items = db.query(Item).offset(pagination.skip).limit(pagination.limit).all()
    total = db.query(Item).count()
    
    return {
        "items": items,
        "page": pagination.page,
        "page_size": pagination.page_size,
        "total": total,
        "pages": (total + pagination.page_size - 1) // pagination.page_size
    }

请求上下文

from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id")

async def set_request_id(request: Request):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request_id_var.set(request_id)
    return request_id

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = await set_request_id(request)
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

依赖调试

import logging

logger = logging.getLogger(__name__)

def debug_dependency(name: str):
    def dependency():
        logger.info(f"Executing dependency: {name}")
        return {"dependency": name}
    return dependency

@app.get("/test")
async def test(
    dep1: dict = Depends(debug_dependency("dep1")),
    dep2: dict = Depends(debug_dependency("dep2"))
):
    # 日志会显示依赖执行顺序
    return {"dep1": dep1, "dep2": dep2}

性能优化：缓存控制

from functools import lru_cache

@lru_cache()
def get_settings():
    # 只读取一次配置文件
    return Settings()

@app.get("/config")
async def get_config(settings: Settings = Depends(get_settings)):
    return settings

返回第二章概览