Skip to main content

Python mediator bus microframework for CQRS + ES

Project description

python-mediator

CI codecov pypi downloads versions license

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)

Uploaded Source

Built Distribution

python_mediator-0.7.0-py3-none-any.whl (28.2 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page