python - 责任链模式实现（2）

pluggy 是 Python 的一个插件管理库。pytest 利用 pluggy 实现插件化，在 Flask 或 Django 等 web 框架中，可以使用 pluggy 为 Middleware 或 View 功能扩展钩子。在数据流管道中，通过 pluggy 添加钩子实现数据采集、过滤、聚合等功能可用于处理数据流的不同阶段。

1. 创建过滤器接口

python">from pluggy import HookimplMarker
from pluggy import HookspecMarker
from pluggy import PluginManagerhookspec = HookspecMarker("example.filter")
hookimpl = HookimplMarker("example.filter")class FilterSpec:"""拦截器接口 ."""@hookspecdef process_request(self, request):pass@hookspecdef process_response(self, request, response):pass@hookspecdef process_view(self, request, view_func, view_args, view_kwargs):pass

FilterSpec 定义了一个插件的接口，其中每个用 @hookspec 装饰的函数可以理解为 hook 钩子 的接口声明。

2. 创建插件即过滤器的实现

python">
class FilterPlugin1:@hookimpldef process_request(self, request):request.data["result"].append("FilterPlugin1")@hookimpl(wrapper=True)def process_response(self, request, response):response.data["data"].append("FilterPlugin1 before yield")# 接受插件 FilterPlugin2 的返回结果results = yieldresults.append("FilterPlugin1")response.data["data"].append("FilterPlugin1 after yield")return resultsclass FilterPlugin2:@hookimpldef process_request(self, request):request.data["result"].append("FilterPlugin2")@hookimpl(wrapper=True)def process_response(self, request, response):response.data["data"].append("FilterPlugin2 before yield")# 接受插件 FilterPlugin3 的返回结果results = yieldresults.append("FilterPlugin2")response.data["data"].append("FilterPlugin2 after yield")return resultsclass FilterPlugin3:@hookimpldef process_request(self, request):request.data["result"].append("FilterPlugin3")@hookimpl(wrapper=True)def process_response(self, request, response):response.data["data"].append("FilterPlugin3 before yield")# 接受其他插件的执行结果results = yieldresults.append("FilterPlugin3")response.data["data"].append("FilterPlugin3 after yield")return results

@hookimpl 装饰器实现了 hook 钩子的业务逻辑。

wrapper=True 可以使得 hook 可以接收其它插件的执行结果。

3. 注册插件

python"># 初始化 PluginManager
pm = PluginManager("example.filter")# 登记 hook 集合(hook函数声明)
pm.add_hookspecs(FilterSpec)# 注册插件(hook函数实现)
pm.register(FilterPlugin3()) # 第一个注册的插件最后执行
pm.register(FilterPlugin2())
pm.register(FilterPlugin1()) # 最后注册的插件最先执行class Request:def __init__(self):self.data = {"result": []}class Response:def __init__(self):self.data = {"code": 0,"data": [],"message": "",}

4. 单元测试

4.1 模拟请求中间件实现

python">def test_hook_request():# 调用两个插件类中的同名hook函数, 后注册的插件中的函数会先被调用request = Request()results = pm.hook.process_request(request=request)assert results == []  # 钩子实现返回 Noneassert request.data == {'result': ['FilterPlugin1', 'FilterPlugin2', 'FilterPlugin3']}

4.2 模拟响应中间件实现

python">def test_hook_response():request = Request()response = Response()results = pm.hook.process_response(request=request, response=response)assert results == ['FilterPlugin3', 'FilterPlugin2', 'FilterPlugin1']assert response.data == {'code': 0,'data': ['FilterPlugin1 before yield','FilterPlugin2 before yield','FilterPlugin3 before yield','FilterPlugin3 after yield','FilterPlugin2 after yield','FilterPlugin1 after yield',],'message': '',}