项目地址: https://github.com/GraiaProject/BroadcastControl
虽然这个项目是我为一个 QQ 机器人框架实现的事件系统, 但即使是单独使用也是没有任何问题的
在 pypi 上找不到一个符合我要求的事件系统(blinker
不支持 asyncio), 并且自己也想借鉴下 FastAPI
的参数获取方式, 就自己手搓了一个.
暂时没有给这个项目做单独文档的精力, 但其稳定性已经在上面那个框架中得到体现.
我设计了一个 Dispatcher 的设计用于规范声明的参数的解析, 单个 /多个的 Dispatcher 都可以作为参数解析器使用(~~突发奇想的一个名称~~):
class Dispatcher(BaseDispatcher):
def catch(self, interface: DispatcherInterface):
if interface.name == "graia" and interface.annotation is str:
return "i'm here."
以上所声明的 Dispatcher, 可以解析这样的 Callable 声明:
def test_execute_target(graia: str):
print(graia) # 输出: "i'm here"
多个 Dispatcher 可以下落式的被调用, 从而可以使 Dispatcher 各司其职.
Dispatcher 还设计了 mixin 模式, 可以引入其他 Dispatcher 的功能.
此外, 事件是这样声明的:
class TestEvent(BaseEvent):
test_field: str
class Dispatcher(BaseDispatcher):
def catch(self, interface):
if interface.name == "test": return interface.event.test_field
只要能实例化事件(这里设实例为 event
), 就能通过 Broadcast.postEvent
方法广播事件:
broadcast.postEvent(event)
事件监听靠 Broadcast.receiver
:
@broadcast.receiver(TestEvent) # 直接使用字符串 "TestEvent" 也是允许的, 会自动查找
async def receiver1(test):
print("当一个 TestEvent 事件被广播时, 最简情况下, 我会被调用", test)
# 同样也支持事件传递阻断: raise PropagationCancelled 即可
# 当返回 RemoveMe 这个特殊的 Signature 时, 可能存在的监听器将被删除.
BaseEvent 的子类都默认继承 pydantic 的 BaseModel, 不允许乱来.
这个事件系统使用 asyncio 的基本 API 实现, 侵入性尽可能的达到了最低. 此外, 还有很多特性, 例如 Decorator
(参数装饰器), Interrupt
(中断) 等没有在这里提到, 有兴趣的人可以在仓库的 src/test.py
处看看, 我也使用 Decorator
实现了类似 FastAPI
的 Depend
和 Middleware
, 但某种程度上更强大.
此外, 你可以在这里阅读我为 Graia Application 写的文档中的对 Broadcast Control 的介绍和基本特性介绍: https://graia-document.vercel.app/docs/broadcast-control/bcc-about
希望能得到意见和支持, 在此感谢大家.
~~怎么没有高亮...~~
1
275761919 2020-12-02 10:16:52 +08:00
可以 很强
|