中间件

中间件介绍

中间件用于在请求到达路由处理函数之前和响应返回客户端之前插入自定义逻辑，适用于全局性的逻辑，如日志、认证、CORS、限流等。

注意：

中间件只处理全局性的逻辑，日志、安全、性能等。不要做业务逻辑，业务逻辑应放在路由函数或服务层。

异常处理要谨慎，中间件中未捕获的异常会导致500错误，建议用 try...except 包裹 call_next。

注册顺序很重要，比如先处理CORS，再记录日志，最后限流。

中间件装饰器

通过 @app.middleware("http") 来装饰中间件函数，适用于实现轻量级逻辑。

该方式简洁、直观，但无法配置参数，难以复用。

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import time

app = FastAPI()

# 定义中间件
@app.middleware("http")
async def add_process_time_header(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

中间件类

通过继承 BaseHTTPMiddleware 来定义中间件类，然后通过 app.add_middleware(中间件类名) 注册中间件到应用。

该方式可以通过__init__传参，易测试、易复用，支持异常处理。

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

class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 前置逻辑(请求到达路由处理函数之前)：记录请求信息
        logging.info(f"{request.method} {request.url}")
        
        # 处理请求：继续处理请求，等待请求处理完再继续
        try:
            response = await call_next(request)
        except Exception as e:
            # 捕获异常并记录
            logging.error(f"Error: {e}")
            raise
        
        # 后置：记录响应状态
        logging.info(f"{response.status_code}")
        return response

# 注册中间件
app = FastAPI()
app.add_middleware(LogMiddleware)

常用内置中间件

中间件	用途	示例
CORSMiddleware	解决跨域问题	几乎所有 Web API 都需要
GZipMiddleware	自动压缩响应	减小传输体积
HTTPSRedirectMiddleware	强制 HTTPS	生产环境安全
TrustedHostMiddleware	限制可信 Host	防止 Host 头攻击

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    # 也可以指定"*"表示允许所有
    allow_origins=["https://your-frontend.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

CORS必须作为第一个中间件注册,否则可能被其他中间件拦截。

依赖注入

依赖项

依赖项指依赖注入时使用的函数，用于执行某些逻辑，可能返回一个值。

Depends

Depends 是 FastAPI 提供的用于声明需要使用依赖注入的类。

依赖注入主要用于复用逻辑、解耦业务代码、自动处理前置条件，提升测试性等。

FastAPI会自动根据依赖函数需要的参数，传入对应需要的形参。

如果路径操作函数是同步函数，则依赖函数只能是同步。如果路径操作函数是异步函数，则依赖函数可以是同步或异步。

注入方式

无返回值依赖注入

from fastapi import Depends, HTTPException

async def verify_admin():
    # 假设从上下文获取角色
    if user_role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")

@app.get("/admin/")
async def admin_panel(_=Depends(verify_admin)):  # 用_表示忽略返回值
    return {"msg": "Welcome, admin!"}

有返回值依赖注入

from typing import Optional
from fastapi import Cookie

async def get_current_user(
    token: Optional[str] = Cookie(
        default=None,              # 没有找到所需Cookie时的默认值
        alias="access_token",      # 必须匹配前端设置的Cookie名
        description="JWT 访问令牌"  # 可选在FastAPI文档中显示说明
    )) -> int:
    if not token:
        raise HTTPException(status_code=401, detail="未登录")
    # 解码并验证token
    payload = jwt.decode(token,
                         SECRET_KEY,
                         algorithms=[ALGORITHM]
                        )
    user_id = payload.get("sub")
    return int(user_id)

@app.get("/dashboard")
async def dashboard(user_id: int = Depends(get_current_user)):
    return {"user_id": user_id}

嵌套依赖注入

一个依赖可以依赖另一个依赖，FastAPI会自动解析依赖链并按顺序调用。

async def get_current_user(token: str = Depends(oauth2_scheme)):
    return decode_jwt(token)

async def get_active_user(current_user=Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(400, "Inactive user")
    return current_user

@app.get("/profile")
async def get_profile(user=Depends(get_active_user)):  # 间接依赖 oauth2_scheme
    return user

作用域

路径操作级依赖

仅对该路径生效，不注入值到函数参数，用于执行检查。

@app.get("/", dependencies=[Depends(verify_token)])
def home():
    ...

路由级依赖

该Router下所有路径都需通过验证。

router = APIRouter(dependencies=[Depends(verify_token)])

应用级全局依赖

所有请求都会执行。

app = FastAPI(dependencies=[Depends(log_requests)])

生命周期

请求生命周期

Middleware：处理所有流经应用的请求/响应。

Depends：只为特定接口提供定制化服务。

客户端请求
    ↓
[Middleware] → 在路由匹配之前执行（不知道去哪）
    ↓
FastAPI 匹配路由（确定调用哪个 endpoint）
    ↓
[Dependencies] → 按依赖链依次执行（已知 endpoint 及其参数）
    ↓
路径操作函数（业务逻辑）
    ↓
[Dependencies cleanup]（如 yield 后的资源释放）
    ↓
[Middleware] → 响应返回前的后处理（如记录耗时）
    ↓
返回响应给客户端

应用生命周期

lifespan (生命周期) 是 FastAPI 用来实现在应用启动前执行初始化操作、在应用关闭后执行清理操作的机制。

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行(Startup)
    # 例如：初始化数据库连接池、加载缓存、启动后台任务等
    print("应用启动！连接数据库...")

    # 应用在此处运行
    yield

    # 关闭时执行(Shutdown)
    # 例如：关闭数据库连接、保存状态、清理资源等
    print("应用关闭！断开数据库...")


# 创建FastAPI实例时传入lifespan
app = FastAPI(lifespan=lifespan)

@app.get("/")
def home():
    return {"status": "running"}