FastAPI 中间件实战：请求日志、耗时统计、IP 黑名单一站式搞定

中间件是什么鬼

中间件是请求处理流程中的核心拦截层。所有传入的HTTP请求和传出的响应都必须经过它，这为统一处理日志、性能监控、安全拦截或注入自定义业务逻辑提供了绝佳位置。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

FastAPI的中间件机制继承自Starlette，其语法极其直观：

@app.middleware("http")
async def my_middleware(request: Request, call_next):
    # 请求进来时做的事
    response = await call_next(request)
    # 请求出去时做的事
    return response

结构清晰明了。call_next是核心函数，它负责将控制权传递给后续的中间件或最终的路由处理器。

实战一：请求日志中间件

记录完整的请求生命周期是运维和调试的基石。你需要准确掌握请求来源、目标路径、响应状态及处理耗时。

import time
from fastapi import Request
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()

    # 记录请求信息
    logger.info(f"→ {request.method} {request.url.path}")

    response = await call_next(request)

    # 计算耗时
    process_time = time.time() - start_time

    # 记录响应信息
    logger.info(f"← {response.status_code} | {process_time:.3f}s")

    return response

部署后，该中间件会为每次API调用生成清晰的审计轨迹。排查500错误或性能瓶颈时，这些日志是首要的定位依据。

实战二：耗时统计中间件

基础的耗时记录可以进一步优化，例如为慢请求添加告警标识，并将性能数据暴露给前端。

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start

    # 超过 1 秒标红
    if elapsed > 1.0:
        logger.warning(f"⚠️ 慢请求: {request.url.path} 耗时 {elapsed:.3f}s")
    else:
        logger.info(f"✓ {request.url.path} 耗时 {elapsed:.3f}s")

    # 把耗时加到响应头里，方便前端调试
    response.headers["X-Process-Time"] = str(elapsed)
    return response

这里使用高精度的perf_counter()进行测量。注入到响应头中的X-Process-Time为前后端协同性能优化提供了直接数据支持。

实战三：IP 黑名单中间件

对于恶意扫描或攻击源IP，在中间件层进行拦截是最有效的防护策略之一。

# 黑名单列表，实际项目中可以放在配置文件或 Redis
BLACKLIST = {"192.168.1.100", "10.0.0.50"}

@app.middleware("http")
async def blacklist_middleware(request: Request, call_next):
    client_ip = request.client.host

    if client_ip in BLACKLIST:
        logger.warning(f"? 拦截黑名单IP: {client_ip}")
        return JSONResponse(
            status_code=403,
            content={"detail": "Access denied"}
        )

    return await call_next(request)

生产环境切忌硬编码IP。应将黑名单动态存储于Redis或数据库，实现实时更新与生效，无需服务重启。

实战四：三合一完整版

将核心功能聚合到一个中间件内，能最大化代码复用并保持处理逻辑的一致性。

import time
from fastapi import Request
from fastapi.responses import JSONResponse
import logging

logger = logging.getLogger(__name__)
BLACKLIST = set()  # 从配置文件读取

@app.middleware("http")
async def unified_middleware(request: Request, call_next):
    # 1. IP 黑名单检查
    client_ip = request.client.host
    if client_ip in BLACKLIST:
        return JSONResponse(status_code=403, content={"detail": "Forbidden"})

    # 2. 开始计时
    start = time.perf_counter()

    # 3. 记录请求
    logger.info(f"[{client_ip}] → {request.method} {request.url.path}")

    # 4. 执行请求
    try:
        response = await call_next(request)
    except Exception as e:
        logger.error(f"? 异常: {e}")
        raise

    # 5. 计算耗时
    elapsed = time.perf_counter() - start

    # 6. 记录响应
    logger.info(f"[{client_ip}] ← {response.status_code} | {elapsed:.3f}s")

    # 7. 添加响应头
    response.headers["X-Process-Time"] = f"{elapsed:.3f}"

    return response

这个增强版中间件集成了IP过滤、结构化日志、异常捕获、性能监控和响应头注入五大功能。单一入口处理全局横切关注点，显著提升代码可维护性。

进阶：多个中间件的执行顺序

FastAPI允许注册多个中间件，其执行顺序遵循“后进先出”的栈模型，形成洋葱式的包裹结构。

# 先注册 A
app.add_middleware(MiddlewareA)
# 再注册 B
app.add_middleware(MiddlewareB)
# 执行顺序：B → A → 路由 → A → B

理解这一顺序对设计中间件依赖关系至关重要。通常，安全类中间件（如IP拦截）应置于最外层，确保尽早拦截非法请求。

写在最后

中间件是FastAPI架构中一个极具威力的组件。虽然路由装饰器也能处理部分逻辑，但中间件的优势在于其全局性和对请求/响应生命周期的完整控制。

它能直接操作原始的Request和Response对象，能力范围覆盖了日志聚合、性能剖析、速率限制、身份验证和CORS配置等众多场景。

在规划API时，应优先识别哪些属于跨切面关注点。将这些公共逻辑抽象到中间件中，能立即提升代码的清晰度与可维护性。

编写健壮的应用不仅关乎功能实现，更在于构建高效的观测与防护体系。一个精心设计的中间件栈，正是实现这一目标的关键基础设施。