Python 设计模式：拦截器 - 指南 - tlnshuju

Python 设计模式：拦截器 - 指南

2025-12-15 19:19 tlnshuju 阅读(0) 评论(0) 收藏举报

Interceptor = 责任链 + AOP 的轻量合体
关键词：钩子、插件、中间件、面向切面编程（AOP）、IoC

引言

当你在一个项目里到处复制粘贴日志、权限校验、限流与审计代码时，是否想过：这些“横切逻辑”为什么不能一次性自动化？拦截器模式提供了一个专业、简洁且可工程化的答案——在既定的连接点上，按顺序挂载可插拔的处理器，把重复工作交给框架，把注意力留给业务。

在这里插入图片描述

你将收获（用户视角）

搭建一个可运行的拦截器骨架（同步/异步）
和装饰器/中间件的差异化理解与选型
可视化的执行流（Mermaid 与 ASCII 图）
可工程化的最佳实践与常见坑位修复

1. 模式定义

拦截器模式（Interceptor / Hook / Plugin）允许在核心处理流程的指定连接点（join-point）上，动态挂接一组可插拔处理器，从而把横切关注点（日志、权限、缓存、限流、审计……）与业务逻辑解耦。

一句话：把“干正事”的代码和“加戏”的代码分开，且“加戏”顺序可配置、可热插拔。

2. 适用场景

场景	举例
Web 框架	Flask/Django 中间件、Starlette/FastAPI 中间件
RPC	gRPC Server/Client Interceptor
任务编排	Airflow Task Hook、Celery Signal
数据管道	ETL 清洗步骤、爬虫请求/响应过滤
客户端 SDK	统一加签、重试、缓存、日志

执行流总览

在这里插入图片描述

3. 最小可运行骨架（同步版）

from typing import Callable, List
from time import perf_counter
# 1. 上下文对象，贯穿整条链
class Context:
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
# 2. 拦截器协议：统一接口
Interceptor = Callable[[Context], None]
# 3. 拦截器链
class InterceptorChain:
def __init__(self):
self._interceptors: List[Interceptor] = []
def add(self, interceptor: Interceptor) -> "InterceptorChain":
self._interceptors.append(interceptor)
return self  # 链式调用
def execute(self, ctx: Context):
for interceptor in self._interceptors:
interceptor(ctx)
# 4. 业务核心（被拦截对象）
def handle_request(ctx: Context):
print(f"[Core] 处理订单 {ctx.order_id}，金额 {ctx.amount}")
# 5. 几个横切拦截器
def log_interceptor(ctx: Context):
print(f"[Log] 开始处理 {ctx.order_id}")
def auth_interceptor(ctx: Context):
if ctx.user_role != "vip":
raise PermissionError("仅限 VIP")
def metrics_interceptor(ctx: Context):
ctx._start = perf_counter()
def post_metrics_interceptor(ctx: Context):
print(f"[Metrics] 耗时 {(perf_counter() - ctx._start)*1000:.2f} ms")
# 6. 装配与运行
if __name__ == "__main__":
chain = (InterceptorChain()
.add(log_interceptor)
.add(metrics_interceptor)
.add(auth_interceptor)
.add(post_metrics_interceptor))
ctx = Context(order_id="OID-123", amount=99.0, user_role="vip")
chain.execute(ctx)   # 先链式前置逻辑
handle_request(ctx)  # 再核心业务

输出示例：

[Log] 开始处理 OID-123
[Metrics] 耗时 0.05 ms
[Core] 处理订单 OID-123，金额 99.0

4. asyncio 异步拦截器

异步场景下，接口签名需改为 async def，并按顺序 await 保证可控的执行链。

import asyncio
from typing import Callable, Awaitable, List
AsyncInterceptor = Callable[[Context], Awaitable[None]]
class AsyncInterceptorChain:
def __init__(self):
self._interceptors: List[AsyncInterceptor] = []
def add(self, interceptor: AsyncInterceptor) -> "AsyncInterceptorChain":
self._interceptors.append(interceptor)
return self
async def execute(self, ctx: Context):
for interceptor in self._interceptors:
await interceptor(ctx)
# 示例异步拦截器
async def rate_limit(ctx: Context):
await asyncio.sleep(0.01)  # 模拟异步 Redis 限流
print("[RateLimit] 通过")
async def main():
chain = AsyncInterceptorChain().add(rate_limit)
await chain.execute(Context(order_id="OID-999"))
asyncio.run(main())

5. 与装饰器、中间件的区别

维度	装饰器	中间件（Django/Flask）	拦截器
作用域	单个函数/类	整个请求/响应周期	任意自定义连接点
顺序控制	硬编码	配置文件	代码/配置皆可
可插拔	需改源码	改配置	改配置
上下文	局部	请求对象	自定义 Context
适合	简单复用	HTTP 横切	通用框架、SDK、数据管道

结论：拦截器 ≈「更通用、更细粒度、可组合」的装饰器/中间件。

6. 工程化最佳实践

接口隔离
用 Protocol 或 ABC 定义拦截器接口，避免直接 Callable 导致类型丢失。

from typing import Protocol
class Interceptor(Protocol):
def intercept(self, ctx: Context) -> None: ...

双向链（前置/后置）
若需要“环绕”逻辑，可让拦截器返回生成器：

def timing():
start = perf_counter()
yield  # 挂起点
print("cost", perf_counter() - start)

异常隔离
给每个拦截器包 try/except，防止单点失败阻断整条链。
热加载
用 importlib.reload 或 entry_points 动态扫描插件目录，实现「上线不停机」。
与依赖注入结合
在 FastAPI 中把拦截器声明为依赖，可自动享受 DI 容器生命周期。

7. 经典源码速览

gRPC Python：grpc.ServerInterceptor / grpc.UnaryUnaryClientInterceptor
Starlette：BaseHTTPMiddleware 就是异步拦截器链
Celery：task_prerun / task_postrun Signal
HTTPX：自定义 httpx.Auth + Event Hooks
建议直接 pip install grpcio starlette celery httpx 后跳转到源码 site-packages 阅读，收获更大。

8. 小结

拦截器 = 责任链 + AOP，用来把“加戏”代码插进主流程。
核心只有三步：定义连接点 → 约定接口 → 顺序执行。
同步用 List[Callable]，异步用 List[Awaitable]；异常隔离、热插拔是工程重点。
Python 世界里，中间件、Signal、Hook、Plugin 基本都是拦截器模式的变体，学会一个就能一通百通。

9. 与 Web 框架的映射

在 FastAPI/Starlette 中：Context ≈ Request/Response，核心处理函数 ≈ call_next，拦截器链可用内置或自定义中间件实现。
在 Koa（Node.js）中：拦截器执行流等价于“洋葱模型”，通过 await next() 控制前置/后置时序。
在 Rust 的 Axum 中：通过 middleware::from_fn 与 Next 构建请求链；复杂场景用 tower::Layer 封装为可复用模块。

洋葱模型（ASCII 可视化）

┌──────────────┐      ┌────────────────────┐      ┌──────────────┐
│ 外层拦截器前 │ ──▶ │   内层拦截器前      │ ──▶ │   核心处理    │
└──────────────┘      └────────────────────┘      └──────────────┘
        ▲                      │                          │
        │                      ▼                          ▼
┌──────────────┐      ┌────────────────────┐      ┌──────────────┐
│ 外层拦截器后 │ ◀── │   内层拦截器后      │ ◀── │   核心返回    │
└──────────────┘      └────────────────────┘      └──────────────┘

随手一跑，就能在控制台看到“加戏”全过程；把 Context 换成 Request/Response，把 handle_request 换成 call_next，你就拥有了属于自己的迷你 Web 框架中间件。

10. 最后想说

当你把重复性的横切工作交给拦截器/中间件，你就能把时间和精力投入到更有创造性的业务设计中去。现在就开始你的自动化之旅吧！

记住：每一个专家都曾经是初学者，区别在于他们开始了。代码示例已经在文中给出，数据就在你的电脑里，剩下的，就是按下那个 Run 按钮。

参考资料

刷新页面返回顶部

tlnshuju