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

fromdataclassesimportdataclassfrommediator.requestimportLocalRequestBusbus=LocalRequestBus()
@dataclassclassPrintMessageCommand:
message: str@bus.registerasyncdefcommand_handler(event: PrintMessageCommand):
print(f"print message: {event.message}")
returnevent.message@dataclassclassDataQuery:
id: int@bus.registerasyncdefquery_handler(query: DataQuery):
print(f"data query: {query.id}")
return {"id": query.id, "data": "test"}
asyncdefmain():
printed_message=awaitbus.execute(PrintMessageCommand(message="test"))
assertprinted_message=="test"data=awaitbus.execute(DataQuery(id=1))
assertdata== {"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

fromdataclassesimportdataclassfrommediator.eventimportLocalEventBusbus=LocalEventBus()
@dataclassclassMessageEvent:
message: str@bus.registerasyncdeffirst_handler(event: MessageEvent):
print(f"first handler: {event.message}")
@bus.registerasyncdefsecond_handler(event: MessageEvent):
print(f"second handler: {event.message}")
asyncdefmain():
awaitbus.publish(MessageEvent(message="test"))
# -- output --# first handler: test# second handler: test

More advanced example available in example/test_event_advanced.py for reference.