Python mediator bus microframework for CQRS + ES
Project description
python-mediator
Elastic and extensible high-performance asyncio CQRS + ES python microframework. Compatible with recent python versions of CPython and pypy3.
Corresponds to clean architecture patterns, ideal for command/query segregation scenarios and event-driven design approaches.
Key features:
- automatic handler inspection and action matching - like in modern frameworks (FastAPI, Typer, Click etc.) machinery is fully automatic and command, query or event object is matched with handler automatically
- extra parameters injection with ease - extra context information like credentials can be passed safely and easily to handler with zero complexity
- configurable middleware (modifier) stack - handler call flow can be extended easily with i.e. data mapping, special exception handling or extra logging by defining modifier stack that wraps handler execution
- ultra-lightweight and performance optimized - has no external dependencies and all features are implemented in care of low runtime overhead
Help
Work in progress...
A command/query handling example
from dataclasses import dataclass
from mediator.request import LocalRequestBus
bus = LocalRequestBus()
@dataclass
class PrintMessageCommand:
message: str
@bus.register
async def command_handler(event: PrintMessageCommand):
print(f"print message: {event.message}")
return event.message
@dataclass
class DataQuery:
id: int
@bus.register
async def query_handler(query: DataQuery):
print(f"data query: {query.id}")
return {"id": query.id, "data": "test"}
async def main():
printed_message = await bus.execute(PrintMessageCommand(message="test"))
assert printed_message == "test"
data = await bus.execute(DataQuery(id=1))
assert data == {"id": 1, "data": "test"}
# -- output --
# print message: test
# data query: 1
More advanced example available in example/test_request_advanced.py for reference.
An event handling example
from dataclasses import dataclass
from mediator.event import LocalEventBus
bus = LocalEventBus()
@dataclass
class MessageEvent:
message: str
@bus.register
async def first_handler(event: MessageEvent):
print(f"first handler: {event.message}")
@bus.register
async def second_handler(event: MessageEvent):
print(f"second handler: {event.message}")
async def main():
await bus.publish(MessageEvent(message="test"))
# -- output --
# first handler: test
# second handler: test
More advanced example available in example/test_event_advanced.py for reference.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
python-mediator-0.7.0.tar.gz
(18.5 kB
view hashes)
Built Distribution
Close
Hashes for python_mediator-0.7.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 1839eb264c724beef24f3d364f9ed8ba73eeddc0b69368df8e886dc09c56ff23 |
|
MD5 | f84f21b854bff530b0f643de64a649ce |
|
BLAKE2b-256 | ac5837406e45d0e21e4ef454cb19cdf1a7726890721eace96d40fe5643b3b8fd |