pythonic命令和事件总线
pybuses的Python项目详细描述
命令和事件总线的pythonic实现
零依赖，灵活实现命令总线。python 3.6+仅
commandbus的基本用法
fromtypingimportListfrompybusesimportCommandBus# Firstly, create command classclassMakeSandwich:def__init__(self,ingredients:List[str])->None:self.ingredients=ingredients# Then create callable responsible for handling command.# It must accept only one argument and is required to have type annotation for it.defsandwich_maker(command:MakeSandwich)->None:print(f'Making sandwich with {command.ingredients}!')# finally, register the handler by subscribingcommand_bus=CommandBus()command_bus.subscribe(sandwich_maker)command_bus.handle(MakeSandwich(['cheese','ham']))
中间产品
中间件是轻量级插件，允许我们在处理给定命令之前和之后注入自定义逻辑。
目前，上下文管理器是定义中间件的唯一受支持的方法。它们简化了异常处理，并指定在处理事件之前还是之后执行中间件逻辑。
importcontextlibfromtypingimport(Any,Generator,)@contextlib.contextmanagerdefexample_middleware(command:Any)->Generator:print(f'Before handling {command}')yieldprint(f'After handling {command}')command_bus=CommandBus([example_middleware])command_bus.subscribe(sandwich_maker)command_bus.handle(MakeSandwich(['cheese','ham']))
eventbus的基本用法
fromdecimalimportDecimalfrompybusesimportEventBus# Create eventclassPaymentMade:amount:Decimalwho:intdef__init__(self,amount:Decimal,who:int)->None:self.amount=amountself.who=whodefhandler(payment_made:PaymentMade)->None:print(f'Oh, cool! {payment_made.who} paid {payment_made.amount / 100}$!')event_bus=EventBus()event_bus.subscribe(handler)event_bus.post(PaymentMade(Decimal('10.99'),123))
事件和命令总线之间的相似性和差异
eventbus可以为每个事件订阅0个或多个处理程序，而commandbus必须为每个命令仅拥有一个处理程序。
CommandBus支持Middleware，而EventBus不支持。
数据类兼容
支持使用dataclasses生成命令/事件
fromdataclassesimportdataclassfromtypingimportListfrompybusesimportCommandBus@dataclassclassMakeSandwich:ingredients:List[str]defhandler(command:MakeSandwich)->None:print(f'dataclass-based command: {command}')command_bus=CommandBus()command_bus.subscribe(handler)command_bus.handle(MakeSandwich(['ham','butter']))
attrs兼容
支持使用attrs生成命令/事件
importattrfrompybusesimportCommandBus@attr.s(frozen=True)classExample:number:int=attr.ib()name:str=attr.ib()defexample_handler(command:Example)->None:print(f'Inside handler of {type(command)} - got {command}!')command_bus=CommandBus()command_bus.subscribe(example_handler)command_bus.handle(Example(number=1,name='Sebastian'))
欢迎加入QQ群-->： 979659372
