A library that implements the Mediator pattern and uses DI library
Project description
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# and follows CQRS principles
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 DiBuilder
Examples
You can find 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 self._user_repo.get_user_by_id(user)
return user
Create DiBuilder
DiBuilder is a facade for Container from DI with caching of solving
di_scopes is a sequence with the order of scopes
di_builder.bind(...) will bind UserRepoImpl type to UserRepo protocol
di_scopes = ("request",)
di_builder = DiBuilder(Container(), AsyncExecutor(), di_scopes)
di_builder.bind(bind_by_type(Dependent(UserRepoImpl, scope="request"), UserRepo))
Create Mediator and register handlers to it
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
dispatchers_middlewares = (LoggingMiddleware(), DiMiddleware(di_builder, cls_scope="request"))
command_dispatcher = CommandDispatcherImpl(middlewares=dispatchers_middlewares)
query_dispatcher = QueryDispatcherImpl(middlewares=dispatchers_middlewares)
mediator = MediatorImpl(command_dispatcher, query_dispatcher)
# 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(...).__call__(...) and inject UserRepoImpl to it
user_id = await scoped_mediator.send(CreateUser(1, "Jon"))
user = await scoped_mediator.query(GetUserById(user_id))
print("User:", user)
# Session of UserRepoImpl will be closed after exiting the "request" scope
⚠️ 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) and .query(query) methods in its Mediator and assumes that you will separate its handlers. Use CommandMediator and QueryMediator protocols to explicitly define which method you need in YourController
graph LR;
YourController-- Command -->Mediator;
YourController-- Query -->Mediator;
Mediator-. Command .->CommandDispatcher-.->di1[DiMiddleware]-.->CommandHandler;
Mediator-. Query .->QueryDispatcher-.->di2[DiMiddleware]-.->QueryHandler;
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
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
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.