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 details)

Uploaded Source

Built Distribution

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

Uploaded Python 3

File details

Details for the file python-mediator-0.7.0.tar.gz.

File metadata

  • Download URL: python-mediator-0.7.0.tar.gz
  • Upload date:
  • Size: 18.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.1.11 CPython/3.8.12 Linux/5.8.0-1042-azure

File hashes

Hashes for python-mediator-0.7.0.tar.gz
Algorithm Hash digest
SHA256 506edc18742cbd6c43eadc654c1f7563c87972b3123e18b2943e1677bb329654
MD5 9e7163df9f68b237c52ca70c1004fc7f
BLAKE2b-256 1264a51a2e55b7fe0189c8ed169ffff52a17a70cb26baf5b535478ffe912c7c7

See more details on using hashes here.

File details

Details for the file python_mediator-0.7.0-py3-none-any.whl.

File metadata

  • Download URL: python_mediator-0.7.0-py3-none-any.whl
  • Upload date:
  • Size: 28.2 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.1.11 CPython/3.8.12 Linux/5.8.0-1042-azure

File hashes

Hashes for python_mediator-0.7.0-py3-none-any.whl
Algorithm Hash digest
SHA256 1839eb264c724beef24f3d364f9ed8ba73eeddc0b69368df8e886dc09c56ff23
MD5 f84f21b854bff530b0f643de64a649ce
BLAKE2b-256 ac5837406e45d0e21e4ef454cb19cdf1a7726890721eace96d40fe5643b3b8fd

See more details on using hashes here.

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