Pythonic Command & Event buses
Project description
Pythonic implementations of Command and Event Buses
Zero-dependencies, flexible implementation of Command Bus. Python 3.6+ only
Basic usage of CommandBus
from typing import List from pybuses import CommandBus # Firstly, create command class class MakeSandwich: 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. def sandwich_maker(command: MakeSandwich) -> None: print(f'Making sandwich with {command.ingredients}!') # finally, register the handler by subscribing command_bus = CommandBus() command_bus.subscribe(sandwich_maker) command_bus.handle(MakeSandwich(['cheese', 'ham']))
Middlewares
Middlewares are lightweight plugins that let us inject custom logic before and after given command was handled.
Context managers are for now the only supported way of defining middlewares. They simplify exception handling and specifying whether middleware logic should be executed before or after handling event.
import contextlib from typing import ( Any, Generator, ) @contextlib.contextmanager def example_middleware(command: Any) -> Generator: print(f'Before handling {command}') yield print(f'After handling {command}') command_bus = CommandBus([example_middleware]) command_bus.subscribe(sandwich_maker) command_bus.handle(MakeSandwich(['cheese', 'ham']))
Basic usage of EventBus
from decimal import Decimal from pybuses import EventBus # Create event class PaymentMade: amount: Decimal who: int def __init__(self, amount: Decimal, who: int) -> None: self.amount = amount self.who = who def handler(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))
Similarities & differences between Event- and CommandBus
EventBus can have 0 or many handlers subscribed to every event, while CommandBus must have exactly one handler for each command.
CommandBus supports middlewares while EventBus does not.
data classes compatible
Using dataclasses for building commands/events is supported
from dataclasses import dataclass from typing import List from pybuses import CommandBus @dataclass class MakeSandwich: ingredients: List[str] def handler(command: MakeSandwich) -> None: print(f'dataclass-based command: {command}') command_bus = CommandBus() command_bus.subscribe(handler) command_bus.handle(MakeSandwich(['ham', 'butter']))
attrs compatible
Using attrs for building commands/events is supported
import attr from pybuses import CommandBus @attr.s(frozen=True) class Example: number: int = attr.ib() name: str = attr.ib() def example_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'))
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.
Filename, size | File type | Python version | Upload date | Hashes |
---|---|---|---|---|
Filename, size pybuses-1.1.0-py3-none-any.whl (7.1 kB) | File type Wheel | Python version py3 | Upload date | Hashes View |
Filename, size pybuses-1.1.0.tar.gz (4.5 kB) | File type Source | Python version None | Upload date | Hashes View |