FastAPI 进阶 — 中间件与依赖注入¶

# pip 安装
pip install "fastapi[standard]"      # 含 uvicorn，当前 0.115+

# 启动开发服务器
uvicorn main:app --reload            # 热重载模式

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# 自定义中间件：记录请求耗时
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()                          # 记录开始时间
        response = await call_next(request)          # 调用下一层（真正的路由）
        duration = time.time() - start               # 计算耗时
        response.headers["X-Process-Time"] = f"{duration:.3f}s"  # 加到响应头
        return response

app.add_middleware(TimingMiddleware)   # 注册中间件

# CORS 中间件（允许前端跨域请求）
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins     = ["http://localhost:3000"],  # 允许的前端域名
    allow_credentials = True,
    allow_methods     = ["*"],                      # 允许所有 HTTP 方法
    allow_headers     = ["*"],
)

from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
import time

app = FastAPI()

# 简单的 IP 限流（生产用 slowapi 或 redis）
request_counts = defaultdict(list)    # IP -> 请求时间戳列表

class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        client_ip  = request.client.host             # 获取客户端 IP
        now        = time.time()
        window     = 60                              # 时间窗口：60秒
        max_reqs   = 100                             # 最多 100 次请求

        # 清除窗口外的记录
        request_counts[client_ip] = [
            t for t in request_counts[client_ip] if now - t < window
        ]

        if len(request_counts[client_ip]) >= max_reqs:
            raise HTTPException(429, "请求太频繁，请稍后重试")

        request_counts[client_ip].append(now)        # 记录本次请求
        return await call_next(request)

app.add_middleware(RateLimitMiddleware)

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Annotated

app = FastAPI()

# 定义依赖函数
def get_db():
    """数据库连接依赖（用完自动关闭）"""
    db = {"connection": "fake_db_connection"}   # 实际用 SQLAlchemy Session
    try:
        yield db                                 # yield 前：建立连接
    finally:
        pass                                     # yield 后：关闭连接（放在 finally）

def verify_token(x_token: Annotated[str, Header()]) -> str:
    """从 Header 验证 API Token"""
    if x_token != "secret-token-123":
        raise HTTPException(401, "Token 无效")
    return x_token

# 在路由中使用依赖
@app.get("/patients/{patient_id}")
async def get_patient(
    patient_id: int,
    db:    Annotated[dict, Depends(get_db)],         # 注入数据库
    token: Annotated[str, Depends(verify_token)],    # 注入鉴权
):
    # 此时 db 和 token 都已准备好
    return {"patient_id": patient_id, "db": db["connection"]}

from fastapi import FastAPI, Depends, Security

# 定义当前用户依赖（从 JWT 解析）
async def get_current_user(token: Annotated[str, Header(alias="authorization")]):
    # 实际应解析 JWT
    if token == "Bearer admin-token":
        return {"user_id": 1, "role": "admin"}
    elif token == "Bearer user-token":
        return {"user_id": 2, "role": "user"}
    raise HTTPException(401, "未登录")

def require_admin(user = Depends(get_current_user)):
    """只允许管理员访问的依赖"""
    if user["role"] != "admin":
        raise HTTPException(403, "权限不足，需要管理员权限")
    return user

@app.delete("/patients/{pid}", dependencies=[Depends(require_admin)])
async def delete_patient(pid: int):
    return {"message": f"患者 {pid} 已删除"}

# 路由组级别应用依赖
from fastapi import APIRouter
admin_router = APIRouter(prefix="/admin", dependencies=[Depends(require_admin)])