samwarden/didiator

Didiator

didiator is an asynchronous library that implements the Mediator pattern and uses the DI library to help you to inject dependencies to called handlers

This library is inspired by the MediatR used in C#, follows CQRS principles and implements event publishing

Installation

Didiator is available on pypi: https://pypi.org/project/didiator

pip install -U "didiator[di]"

It will install didiator with its optional DI dependency that is necessary to use DiMiddleware and DiBuilderImpl

Examples

You can find more examples in this folder

Create Commands and Queries with handlers for them

@dataclass
class CreateUser(Command[int]):
    user_id: int
    username: str

class CreateUserHandler(CommandHandler[CreateUser, int]):
    def __init__(self, user_repo: UserRepo) -> None:
        self._user_repo = user_repo

    async def __call__(self, command: CreateUser) -> int:
        user = User(id=command.user_id, username=command.username)
        await self._user_repo.add_user(user)
        await self._user_repo.commit()
        return user.id

You can use functions as handlers

@dataclass
class GetUserById(Query[User]):
    user_id: int

async def handle_get_user_by_id(query: GetUserById, user_repo: UserRepo) -> User:
    user = await user_repo.get_user_by_id(query.user_id)
    return user

Create DiBuilder

DiBuilderImpl is a facade for Container from DI with caching of solving

di_scopes is a list with the order of scopes

di_builder.bind(...) will bind UserRepoImpl type to UserRepo protocol

di_scopes = ["request"]
di_builder = DiBuilderImpl(Container(), AsyncExecutor(), di_scopes)
di_builder.bind(bind_by_type(Dependent(UserRepoImpl, scope="request"), UserRepo))

Create Mediator

Create dispatchers with their middlewares and use them to initialize the MediatorImpl

cls_scope is a scope that will be used to bind class Command/Query handlers initialized during request handling

middlewares = (LoggingMiddleware(), DiMiddleware(di_builder, scopes=DiScopes("request")))
command_dispatcher = CommandDispatcherImpl(middlewares=middlewares)
query_dispatcher = QueryDispatcherImpl(middlewares=middlewares)

mediator = MediatorImpl(command_dispatcher, query_dispatcher)

Register handlers

# CreateUserHandler is not initialized during registration
mediator.register_command_handler(CreateUser, CreateUserHandler)
mediator.register_query_handler(GetUserById, handle_get_user_by_id)

Main usage

Enter the "request" scope that was registered earlier and create a new Mediator with di_state bound

Use mediator.send(...) for commands and mediator.query(...) for queries

async with di_builder.enter_scope("request") as di_state:
    scoped_mediator = mediator.bind(di_state=di_state)

    # It will call CreateUserHandler(UserRepoImpl()).__call__(command)
    # UserRepoImpl() created and injected automatically
    user_id = await scoped_mediator.send(CreateUser(1, "Jon"))

    # It will call handle_get_user_by_id(query, user_repo)
    # UserRepoImpl created earlier will be reused in this scope
    user = await scoped_mediator.query(GetUserById(user_id))
    print("User:",  user)
# Session of UserRepoImpl will be closed after exiting the "request" scope

Events publishing

You can register and publish events using Mediator and its EventObserver. Unlike dispatchers, EventObserver publishes events to multiple event handlers subscribed to it and doesn't return their result. All middlewares also work with EventObserver, as in in the case with Dispatchers.

Define event and its handlers

class UserCreated(Event):
    user_id: int
    username: str

async def on_user_created1(event: UserCreated, logger: Logger) -> None:
    logger.info("User created1: id=%s,  username=%s", event.user_id, event.username)

async def on_user_created2(event: UserCreated, logger: Logger) -> None:
    logger.info("User created2: id=%s,  username=%s", event.user_id, event.username)

Create EventObserver and use it for Mediator

middlewares = (LoggingMiddleware(), DiMiddleware(di_builder, scopes=DiScopes("request")))
event_observer = EventObserver(middlewares=middlewares)

mediator = MediatorImpl(command_dispatcher, query_dispatcher, event_observer)

Register event handlers

You can register multiple event handlers for one event

mediator.register_event_handler(UserCreated, on_user_created1)
mediator.register_event_handler(UserCreated, on_user_created2)

Publish event

Event handlers will be executed sequentially

await mediator.publish(UserCreated(1, "Jon"))
# User created1: id=1,  username="Jon"
# User created2: id=1,  username="Jon"

await mediator.publish([UserCreated(2, "Sam"), UserCreated(3, "Nick")])
# User created1: id=2,  username="Sam"
# User created2: id=2,  username="Sam"
# User created1: id=3,  username="Nick"
# User created2: id=3,  username="Nick"

⚠️ Attention: this is a beta version of didiator that depends on DI, which is also in beta. Both of them can change their API!

CQRS

CQRS stands for "Command Query Responsibility Segregation". Its idea about splitting the responsibility of commands (writing) and queries (reading) into different models.

didiator have segregated .send(command), .query(query) and .publish(events) methods in its Mediator and assumes that you will separate its handlers. Use CommandMediator, QueryMediator and EventMediator protocols to explicitly define which method you need in YourController

graph LR;
    YourController-- Query -->Mediator;
    YourController-- Command -->Mediator;
    Mediator-. Query .->QueryDispatcher-.->di2[DiMiddleware]-.->QueryHandler;
    Mediator-. Command .->CommandDispatcher-.->di1[DiMiddleware]-.->CommandHandler;
    CommandHandler-- Event -->Mediator;
    Mediator-. Event .->EventObserver-.->di3[DiMiddleware]-.->EventHandler1;
    EventObserver-.->di4[DiMiddleware]-.->EventHandler2;

DiMiddleware initializes handlers and injects dependencies for them, you can just send a command with the data you need

Why didiator?

• Easy dependency injection to your business logic
• Separating dependencies from your controllers. They can just parse external requests and interact with the Mediator
• CQRS
• Event publishing
• Flexible configuration
• Middlewares support

Why not?

• You don't need it
• Maybe too low coupling: navigation becomes more difficult
• Didiator is in beta now
• No support for synchronous handlers

didiator's Issues

Should didiator have EventObserver for publishing asynchronous events for several handlers?

Add logging for dispatchers

Add logging to CommandDispatcherImpl and QueryDispatcherImpl. Both should have diferrent description for logs.

Alternatively LoggingMiddleware can be added

Refactor handler wrapping by middlewares to have an opportunity call it without necessary explictly initialize it

Add middleware for asynchronous handling of commands

Add AsyncHandlingMiddleware that will accept is_async parameter from extra data and if it's True, then create a task for a next handler and return None without awaiting its result

Add copying to Dispatchers and Observer

Add method .copy() to Dispatchers and Observer to create a new class with the same handlers

Add linters and auto-tests

Didiator developed and tested with using python3.10, but has to work with python3.11.
Configure tox and use it to run auto-tests with different python versions

Add sync didiator support

Fix typehints issues

Add benchmarks

Compare performance of the Mediator and an example implementation of a Service

Add a way to inject additional dependencies from middlewares

Add a way to inject additional dependencies from web framework middlewares during initialization a scoped mediator.
For example

scoped_mediator = mediator.bind(di_state=di_state, di_values={AccessPolicy: AccessPolicy()})

Implement `dishka` package integration