一个现代的事件系统
Project description
EventGraph
EventGraph 是一个现代的事件系统,旨在为 Python 应用提供高效、灵活的事件处理机制。
特性
- 类型安全:利用 Python 的类型提示来确保事件处理的安全性。
- 异步支持:原生支持异步事件处理。
- 易于扩展:面向接口编程,易于扩展新的事件处理器。
- 上下文切换:支持上下文切换,方便在事件处理中传递数据。
- 消息队列:支持消息队列,意味着可以跨进程、跨机器进行事件处理。
快速开始
安装
使用 pip 安装 EventGraph:
pip install eventgraph
基本用法
- 初始化事件图:
import asyncio
from eventgraph.core.core import EventGraph, init_event_graph
from eventgraph.dispatcher.base import Dispatcher
from eventgraph.context import InstanceContext
from eventgraph.exceptions import NoCatchArgs
g = init_event_graph(int, InstanceContext())
- 定义事件处理函数:
@g.receiver(int)
async def test1(a: int, b: str, c=1):
print(locals(), "test1")
@g.receiver(Ts)
async def test2(a: Ts, b: str, c=1, d: Optional[EventGraph] = None):
print(locals(), "test2")
- 定义参数处理器:
class IntDispatcher(Dispatcher[EventGraph[int], int]):
@classmethod
async def catch(cls, interface):
if interface.annotation == str:
return "string"
raise NoCatchArgs
g.add_dispatcher(int, IntDispatcher)
- 发布和执行事件:
async def mian():
g.start()
g.postEvent(1) # 发布事件
g.postEvent(Ts(2))
await g.execute(Ts(1)) # 直接执行事件
await asyncio.sleep(1)
asyncio.run(mian())
鸣谢
没有它们的存在就没有这个项目
项目灵感
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
eventgraph-0.1.0rc3.tar.gz
(12.7 kB
view hashes)
Built Distribution
Close
Hashes for eventgraph-0.1.0rc3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | b3ae842073814850c91fceb3269d3b1285ff58d9462b4cd99c7cf982617a7941 |
|
MD5 | bcfa8de683cdccb0d329ef63567604a4 |
|
BLAKE2b-256 | 9e1c843cca5ed787aa804e1dbba1cdc6fd458889d38142adf6593df43f4a01c9 |