1 实现一个类似Flask事件插件

我们知道flask的框架有一些钩子事件

请求前（before_request）
请求后（after_request）
请求结束后（teardown_appcontext）

参考信号库的源码，于是乎我们可以自定义实现类似的信号事件处理，并结合中间件来实现类似的钩子函数。

目前新脚手架自定义一些插件库，如下所示：

2 自定义信号库的实现blinker

首先定义一个存贮字典，用于保持对于的注册的事件信息，如下代码所示：

#!/usr/bin/evn python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   文件名称 :     __init__.py
   文件功能描述 :   功能描述
   创建人 :       小钟同学
   创建时间 :          2021/12/22
-------------------------------------------------
   修改描述-2021/12/22:  参考信号库自己仿写的一个事件处理机制，
-------------------------------------------------

"""

from collections import defaultdict
import inspect
import functools
from framecore.libs.transf_sync_aysnc.sync_to_async import threadpool_run_in_executor



def hashable_identity(obj):
    if hasattr(obj, '__func__'):
        return (id(obj.__func__), id(obj.__self__))
    elif hasattr(obj, 'im_func'):
        return (id(obj.im_func), id(obj.im_self))
    elif isinstance(obj, str):
        return obj
    else:
        return id(obj)


class Event:
    pass

    def __init__(self):
        # 定义所有的注册事件对象信息，使用一个字典来保存，对应的事件的ID和事件注册的函数或对象-所有的订阅对象
        self.receivers_subscribes = {}
        # 保持的是注册的对象事件源ID 和 接受者对象的名称
        self._by_receiver = defaultdict(set)  # 创建集合set---默认的是一个字典，当找不到对应的key返回的一个set类型的对象，而不是会报空
        # 保存的是所有的发送者（类似主题类型的分配）
        self._by_sender = defaultdict(set)  # 对应的一个发送者所有的书画报

    def register_subscribe(self, receiver, sender='ZYXEvent'):
        '''
         开始创建一个方法，用于把订阅对象信息进行存贮并根据相关的发生事件的名称和主题等分类
         ：PS:一个事件可以有多个主题，或多个发送至，可以根据主题和发送者过滤相关的处理
        :param receiver: 注册对象（订阅者函数或类或其他）
        :param sender: 发送者（或发送主题名称）
        :return:
        '''
        pass
        # 创建一个注册对象的唯一的ID索引，用户标记定义这的唯一信息
        # 生成订阅者对象ID
        receiver_id = hashable_identity(receiver)
        # 生成发送者的ID
        sender_id = hashable_identity(sender)
        # 把订阅者保持到字典里面进行存贮
        self.receivers_subscribes.setdefault(receiver_id, receiver)
        # 删除当前注册的订阅者，减少引用
        # 记录不同的发送者对应的多个的发送者的ID信息：{'sdfsdf': {'nihao1', 'nihao3', 'nihao2'}, 'sdfsdf2': {'nihao3'}})
        # 一个发送者可以对应多个注册者的信息
        self._by_sender[sender_id].add(receiver_id)
        # 记录每订阅者对象ID，内都有哪些发送者的ID信息---用于过滤且当且只有存在这个发送者信息的时候，才会进行事件回调处理
        self._by_receiver[receiver_id].add(sender_id)
        # 范湖idangq的注册者信息
        return receiver

    def register_subscribe_decorator(self, sender):
        '''
        有惨的时候，需要传参，不能使用默认值，且需要此装饰器
        :param sender:
        :return:
        '''

        def decorator(fn):
            self.register_subscribe(fn, sender)
            return fn

        return decorator

    async def send_async(self, sender='ZYXEvent', **kwargs):

        if not self.receivers_subscribes:
            return []
        else:
            iscoroutinefunctions=[]
            for receiver in self.receivers_for(sender):
                if inspect.iscoroutinefunction(receiver):
                    # 添加的协程的列表里面用户后续的批量的执行
                    #将协程转成task，并组成list
                    # print("应该是协程啊！！！",asyncio.create_task(receiver))
                    # iscoroutinefunctions.append(receiver)
                    await receiver(sender, **kwargs)
                else:
                    # 把同步的阻塞的任务或函数的执行，转为异步可以等待的协程的任务
                    # 但是本质主要作用不是把同步变成协程运行，而是让其拥有了异步await 的用法，
                    # 既能不阻塞当前事件循环，又能在同步函数执行完成return结果时拿到结果接着用
                    # loop = asyncio.get_event_loop()
                    # threadpool_executor = None or global_threadpool_executor
                    # await loop.run_in_executor(executor=threadpool_executor,func=functools.partial(receiver, sender, **kwargs))
                    await threadpool_run_in_executor(func=functools.partial(receiver, sender, **kwargs))


    def send_sync(self, sender='ZYXEvent', **kwargs):

        if not self.receivers_subscribes:
            return []
        else:
            return [receiver(sender, **kwargs) for receiver in self.receivers_for(sender)]

    def receivers_for(self, sender):
        """Iterate all live receivers listening for *sender*."""
        pass
        if self.receivers_subscribes:
            pass
            # 根据发送者ID--获取对应的对应多个注册者的信息ids
            sender_id = hashable_identity(sender)
            # 根据发生者ID来 获取相关的订阅者的IDS信息
            receiver_ids = (self._by_sender[sender_id])
            if not receiver_ids:
                return

            for receiver_id in receiver_ids:
                receiver = self.receivers_subscribes.get(receiver_id)
                if receiver is None:
                    continue
                yield receiver

    def has_receivers_for(self, sender):

        if not self.receivers_subscribes:
            return False
        if self._by_sender[sender]:
            return True
        return hashable_identity(sender) in self._by_sender


class NamedEvent(Event):
    """A named generic notification emitter."""

    def __init__(self, name, ):
        Event.__init__(self)

        #: The name of this signal.
        self.name = name

    def __repr__(self):
        base = Event.__repr__(self)
        return "%s; %r>" % (base[:-1], self.name)


class NameEventspace(dict):
    """A mapping of signal names to signals."""

    def event(self, name):
        """Return the :class:`NamedSignal` *name*, creating it if required.
        Repeated calls to this function will return the same signal object.
        """
        try:
            return self[name]
        except KeyError:
            return self.setdefault(name, NamedEvent(name))

    def has_receiver(self, name):
        """Return the :class:`NamedSignal` *name*, creating it if required.
        Repeated calls to this function will return the same signal object.
        """
        return name in self

使用示例如下:

#!/usr/bin/evn python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   文件名称 :     event
   文件功能描述 :   功能描述
   创建人 :       小钟同学
   创建时间 :          2021/12/23
-------------------------------------------------
   修改描述-2021/12/23:

-------------------------------------------------
"""

from framecore.libs.evnets_pcb import NameEventspace, NamedEvent

# Namespace()-里面的一个方法signal--用于注册事件名称，存如到字典里面
# 默认存贮的是：NamedSignal--->Signal

if __name__ == '__main__':
    event = NameEventspace().event
    name_signal1: NamedEvent = event("你好吗！！")  # 这里根据name创建一个信号
    name_signal2: NamedEvent = event("你好吗！！")  # 这里根据name创建一个信号
    print(name_signal1.name)
    print(name_signal2.__repr__)

    name_signal: NamedEvent = event("你好吗！！")  # 这里其实没有创建新的信号，就是之前的name信号\


    @name_signal.register_subscribe_decorator(sender="N")  # 绑定了订阅者
    def wx_notify_third_inters_error1(sender, **kw):
        print("接收到了1！！！！！！！！！", sender, kw)


    #
    @name_signal.register_subscribe_decorator(sender="NA")  # 绑定了订阅者
    def wx_notify_third_inters_error2(sender, **kw):
        print("接收到了22222222222222！！！！！！！！！", sender, kw)


    ret = name_signal1.send(sender="NA", source={
        "nis": 343452,
        "nis222222222": '你喜欢我吗'
    }, traceback='你喜欢我吗')  # 发送一个信号

    name_signal1: NamedEvent = event("你好吗2！！")  # 这里根据name创建一个信号


    @name_signal1.register_subscribe_decorator(sender="N")  # 绑定了订阅者
    def wx_notify_third_inters_error1(sender, **kw):
        print("接收到了1！！！！！！！！！", sender, kw)


    #
    @name_signal1.register_subscribe_decorator(sender="NA")  # 绑定了订阅者
    def wx_notify_third_inters_error2(sender, **kw):
        print("接收到了22222222222222！！！！！！！！！", sender, kw)


    name_signal1.send_sync(sender="NA", source={
        "nis": 343452,
        "nis222222222": '你喜欢我吗'
    }, traceback='你喜欢我吗')  # 发送一个信号

    name_signal1: NamedEvent = event("你好吗！！")  # 这里根据name创建一个信号
    name_signal2: NamedEvent = event("你好吗！！")  # 这里根据name创建一个信号
    print(id(name_signal1),id(name_signal2))

关于代码说明有具体的注释备注，这里我不多解析。

3然后到我们的插件定义

完整源码,如下示例：

from framecore.plugins.apluginbase import PluginBase, PluginBaseSettings
from framecore.libs.evnets_pcb import NameEventspace
import typing
from fastapi import FastAPI


class RequestEventHookSettings(PluginBaseSettings):
    # allow_origins = ["*"]
    pass


class RequestEventHookPluginClient(PluginBase):
    # 设置插件默认的参数信息
    defualt_config = RequestEventHookSettings

    # 自定义全局的时间命名空间
    eventspace = NameEventspace()
    is_open_request_hook = True

    def add_register_event_handler(self, func: typing.Callable, eventname: str, sender='FastapiEvent',
                                   onlyonehandler=False) -> typing.Callable:
        '''
        允许重复注册多个处理器回调
        :param func:
        :param eventname:
        :param sender:
        :return:
        '''
        if eventname and onlyonehandler and not self.eventspace.has_receiver(eventname):
            # 如果只允许注册一个处理器对象的话就需要过滤避免多重注册
            ememt = self.eventspace.event(eventname)
            ememt.register_subscribe(func, sender)
        else:
            #  允许重复注册多个处理器回调
            ememt = self.eventspace.event(eventname)
            ememt.register_subscribe(func, sender)
        return func

    def before_request(self, func: typing.Callable, eventname: str = 'before_request',
                       sender='FastapiEvent') -> typing.Callable:
        if self.is_open_request_hook:
            ememt = self.eventspace.event(eventname)
            ememt.register_subscribe(func, sender)
        return func

    def after_request(self, func: typing.Callable, eventname: str = 'after_request',
                      sender='FastapiEvent') -> typing.Callable:
        if self.is_open_request_hook:
            ememt = self.eventspace.event(eventname)
            ememt.register_subscribe(func, sender)
        return func

    def teardown_appcontext(self, func: typing.Callable, eventname: str = 'teardown_appcontext',
                            sender='FastapiEvent') -> typing.Callable:
        if self.is_open_request_hook:
            ememt = self.eventspace.event(eventname)
            ememt.register_subscribe(func, sender)
        return func

    async def send_async_event(self, eventname, sender='FastapiEvent', **kwargs):
        '''
        有惨的时候，需要传参，不能使用默认值，且需要此装饰器
        :param sender:
        :return:
        '''
        enent = self.eventspace
        # nonlocal enent
        if eventname and enent.has_receiver(eventname):
            pass
            ememt = enent.event(eventname)
            await ememt.send_async(sender, **kwargs)

    def send_sync_event(self, eventname, sender='ZYXEvent', **kwargs):
        '''
        有惨的时候，需要传参，不能使用默认值，且需要此装饰器
        :param sender:
        :return:
        '''
        enent = self.eventspace
        # nonlocal enent
        if eventname and enent.has_receiver(eventname):
            pass
            ememt = enent.event(eventname)
            ememt.send_sync(sender, **kwargs)

    def register_event(self, eventname: str, sender='FastapiEvent') -> typing.Callable:
        enent = self.eventspace
        def decorator(func: typing.Callable) -> typing.Callable:
            if eventname and not enent.has_receiver(eventname):
                ememt = enent.event(eventname)
                ememt.register_subscribe(func, sender)
            return func

        return decorator

    def mount_app(self, app: FastAPI, config: RequestEventHookSettings = None, *args, **kwargs):
        thism= self
        @app.middleware("http")
        async def event_request(request, call_next):
            response = None
            nonlocal thism
            try:
                await thism.send_async_event(eventname='before_request', sender='FastapiEvent', request=request)
                response = await call_next(request)
                await thism.send_async_event(eventname='after_request', sender='FastapiEvent', response=response)
                return response
            finally:
                pass
                await thism.send_async_event(eventname='teardown_appcontext', sender='FastapiEvent', response=response)
        return self

代码解析说明

eventspace = NameEventspace()创建一个事件处理命名空间对象
通过add_register_event_handler进行相关的事件注册，其中参数主要是定义事件的名称和具体的事件的生产者，也可以通过装饰器register_event来机芯事件的注册。
通过send_async_event获取到具体的事件并执行发送处理
通过mount_app挂载传递一个app对象过来，并直接进行当前事件分发中间件的注册处理
在中间件中，我们在对于的位置对当前相关的事件名称和事件的发送信息进行相关的事件的处理发送。

使用方式如下：

from fastapi import FastAPI
app = FastAPI(docs_url=None,redoc_url=None)

from framecore.plugins.healthchcke import HealthCheckPluginClient
HealthCheckPluginClient().mount_app(app)

from framecore.plugins.cors import CorsPluginClient
CorsPluginClient().mount_app(app)

from framecore.plugins.swaggerui import SwaggeruiPluginClient
SwaggeruiPluginClient().mount_app(app)

from framecore.plugins.reuqest_event_hook import RequestEventHookPluginClient
name_signal:RequestEventHookPluginClient = RequestEventHookPluginClient().mount_app(app)
@name_signal.register_event(eventname='before_request')  # 绑定了订阅者
def wx_notify_third_inters_error1(sender, **kw):
    print("接收到了1！！！！！！！！！", sender, kw)



@app.get("/")
def asdas():
    return ''

if __name__ == "__main__":
    from framecore.utils.routes_helper import print_all_routes_info
    print_all_routes_info(app)
    import uvicorn
    import os
    app_modeel_name = os.path.basename(__file__).replace(".py", "")
    print(app_modeel_name)
    uvicorn.run(f"{app_modeel_name}:app", host='127.0.0.1',port=31100,reload=True,workers=3)

关键代码是：


from framecore.plugins.reuqest_event_hook import RequestEventHookPluginClient
name_signal:RequestEventHookPluginClient = RequestEventHookPluginClient().mount_app(app)
@name_signal.register_event(eventname='before_request')  # 绑定了订阅者
def wx_notify_third_inters_error1(sender, **kw):
    print("接收到了1！！！！！！！！！", sender, kw)

通过上面进行相关事件注册，当我们的请求开始进行处理具体的before_request经常的时候，则会执行具体的回调处理。这样可以分离业务，便于业务解耦！

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文【欢迎一起学习交流】| QQ：308711822

ITPub博客

fastapi框架系列-插件系列（1）-实现类似flask钩子事件的插件

1 实现一个类似Flask事件插件

2 自定义信号库的实现blinker

3然后到我们的插件定义

代码解析说明

结尾

END

最新文章