========
Didiator
========
``didiator`` is an asynchronous library that implements the Mediator pattern and
uses the `DI <https://www.adriangb.com/di/>`_ library to help you to inject dependencies to called handlers
This library is inspired by the `MediatR <https://github.com/jbogard/MediatR>`_ used in C#,
follows CQRS principles and implements event publishing
Installation
============
Didiator is available on pypi: https://pypi.org/project/didiator
.. code-block:: bash
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 <https://github.com/SamWarden/didiator/tree/dev/examples>`_
Create Commands and Queries with handlers for them
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
@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
.. code-block:: python
@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 <https://www.adriangb.com/di/0.73.0/solving/>`_
``di_scopes`` is a list with the order of `scopes <https://www.adriangb.com/di/0.73.0/scopes/>`_
``di_builder.bind(...)`` will `bind <https://www.adriangb.com/di/0.73.0/binds/>`_ ``UserRepoImpl`` type to ``UserRepo`` protocol
.. code-block:: python
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
.. code-block:: python
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
~~~~~~~~~~~~~~~~~
.. code-block:: python
# 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
.. code-block:: python
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
-----------------------------
.. code-block:: python
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
--------------------------------------------
.. code-block:: python
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
.. code-block:: python
mediator.register_event_handler(UserCreated, on_user_created1)
mediator.register_event_handler(UserCreated, on_user_created2)
Publish event
-------------
Event handlers will be executed sequentially
.. code-block:: python
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 <https://www.martinfowler.com/bliki/CQRS.html>`_".
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``
.. code-block:: mermaid
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
Raw data
{
"_id": null,
"home_page": "https://github.com/SamWarden/didiator",
"name": "didiator",
"maintainer": "SamWarden",
"docs_url": null,
"requires_python": ">=3.10,<4.0",
"maintainer_email": "SamWardenSad@gmail.com",
"keywords": "didiator,mediatr,mediator,CQRS,DI,events,ioc",
"author": "SamWarden",
"author_email": "SamWardenSad@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/20/d7/8f577ae09ed6ee6b5df6cb669e34d7cc848c80a7a2045b3e9f1921c80ded/didiator-0.3.0.tar.gz",
"platform": null,
"description": "========\nDidiator\n========\n\n``didiator`` is an asynchronous library that implements the Mediator pattern and\nuses the `DI <https://www.adriangb.com/di/>`_ library to help you to inject dependencies to called handlers\n\nThis library is inspired by the `MediatR <https://github.com/jbogard/MediatR>`_ used in C#,\nfollows CQRS principles and implements event publishing\n\nInstallation\n============\n\nDidiator is available on pypi: https://pypi.org/project/didiator\n\n.. code-block:: bash\n\n pip install -U \"didiator[di]\"\n\nIt will install ``didiator`` with its optional DI dependency that is necessary to use ``DiMiddleware`` and ``DiBuilderImpl``\n\nExamples\n========\n\nYou can find more examples in `this folder <https://github.com/SamWarden/didiator/tree/dev/examples>`_\n\nCreate Commands and Queries with handlers for them\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n\n.. code-block:: python\n\n @dataclass\n class CreateUser(Command[int]):\n user_id: int\n username: str\n\n class CreateUserHandler(CommandHandler[CreateUser, int]):\n def __init__(self, user_repo: UserRepo) -> None:\n self._user_repo = user_repo\n\n async def __call__(self, command: CreateUser) -> int:\n user = User(id=command.user_id, username=command.username)\n await self._user_repo.add_user(user)\n await self._user_repo.commit()\n return user.id\n\nYou can use functions as handlers\n\n.. code-block:: python\n\n @dataclass\n class GetUserById(Query[User]):\n user_id: int\n\n async def handle_get_user_by_id(query: GetUserById, user_repo: UserRepo) -> User:\n user = await user_repo.get_user_by_id(query.user_id)\n return user\n\nCreate DiBuilder\n~~~~~~~~~~~~~~~~\n\n``DiBuilderImpl`` is a facade for Container from DI with caching of `solving <https://www.adriangb.com/di/0.73.0/solving/>`_\n\n``di_scopes`` is a list with the order of `scopes <https://www.adriangb.com/di/0.73.0/scopes/>`_\n\n``di_builder.bind(...)`` will `bind <https://www.adriangb.com/di/0.73.0/binds/>`_ ``UserRepoImpl`` type to ``UserRepo`` protocol\n\n.. code-block:: python\n\n di_scopes = [\"request\"]\n di_builder = DiBuilderImpl(Container(), AsyncExecutor(), di_scopes)\n di_builder.bind(bind_by_type(Dependent(UserRepoImpl, scope=\"request\"), UserRepo))\n\nCreate Mediator\n~~~~~~~~~~~~~~~\n\nCreate dispatchers with their middlewares and use them to initialize the ``MediatorImpl``\n\n``cls_scope`` is a scope that will be used to bind class Command/Query handlers initialized during request handling\n\n.. code-block:: python\n\n middlewares = (LoggingMiddleware(), DiMiddleware(di_builder, scopes=DiScopes(\"request\")))\n command_dispatcher = CommandDispatcherImpl(middlewares=middlewares)\n query_dispatcher = QueryDispatcherImpl(middlewares=middlewares)\n\n mediator = MediatorImpl(command_dispatcher, query_dispatcher)\n\nRegister handlers\n~~~~~~~~~~~~~~~~~\n\n.. code-block:: python\n\n # CreateUserHandler is not initialized during registration\n mediator.register_command_handler(CreateUser, CreateUserHandler)\n mediator.register_query_handler(GetUserById, handle_get_user_by_id)\n\nMain usage\n~~~~~~~~~~\n\nEnter the ``\"request\"`` scope that was registered earlier and create a new Mediator with ``di_state`` bound\n\nUse ``mediator.send(...)`` for commands and ``mediator.query(...)`` for queries\n\n.. code-block:: python\n\n async with di_builder.enter_scope(\"request\") as di_state:\n scoped_mediator = mediator.bind(di_state=di_state)\n\n # It will call CreateUserHandler(UserRepoImpl()).__call__(command)\n # UserRepoImpl() created and injected automatically\n user_id = await scoped_mediator.send(CreateUser(1, \"Jon\"))\n\n # It will call handle_get_user_by_id(query, user_repo)\n # UserRepoImpl created earlier will be reused in this scope\n user = await scoped_mediator.query(GetUserById(user_id))\n print(\"User:\", user)\n # Session of UserRepoImpl will be closed after exiting the \"request\" scope\n\nEvents publishing\n~~~~~~~~~~~~~~~~~\n\nYou can register and publish events using ``Mediator`` and its ``EventObserver``.\nUnlike dispatchers, ``EventObserver`` publishes events to multiple event handlers subscribed to it\nand doesn't return their result.\nAll middlewares also work with ``EventObserver``, as in in the case with Dispatchers.\n\nDefine event and its handlers\n-----------------------------\n\n.. code-block:: python\n\n class UserCreated(Event):\n user_id: int\n username: str\n\n async def on_user_created1(event: UserCreated, logger: Logger) -> None:\n logger.info(\"User created1: id=%s, username=%s\", event.user_id, event.username)\n\n async def on_user_created2(event: UserCreated, logger: Logger) -> None:\n logger.info(\"User created2: id=%s, username=%s\", event.user_id, event.username)\n\nCreate EventObserver and use it for Mediator\n--------------------------------------------\n\n.. code-block:: python\n\n middlewares = (LoggingMiddleware(), DiMiddleware(di_builder, scopes=DiScopes(\"request\")))\n event_observer = EventObserver(middlewares=middlewares)\n\n mediator = MediatorImpl(command_dispatcher, query_dispatcher, event_observer)\n\nRegister event handlers\n-----------------------\n\nYou can register multiple event handlers for one event\n\n.. code-block:: python\n\n mediator.register_event_handler(UserCreated, on_user_created1)\n mediator.register_event_handler(UserCreated, on_user_created2)\n\nPublish event\n-------------\n\nEvent handlers will be executed sequentially\n\n.. code-block:: python\n\n await mediator.publish(UserCreated(1, \"Jon\"))\n # User created1: id=1, username=\"Jon\"\n # User created2: id=1, username=\"Jon\"\n\n await mediator.publish([UserCreated(2, \"Sam\"), UserCreated(3, \"Nick\")])\n # User created1: id=2, username=\"Sam\"\n # User created2: id=2, username=\"Sam\"\n # User created1: id=3, username=\"Nick\"\n # User created2: id=3, username=\"Nick\"\n\n\u26a0\ufe0f **Attention: this is a beta version of** ``didiator`` **that depends on** ``DI``, **which is also in beta. Both of them can change their API!**\n\nCQRS\n====\n\nCQRS stands for \"`Command Query Responsibility Segregation <https://www.martinfowler.com/bliki/CQRS.html>`_\".\nIts idea about splitting the responsibility of commands (writing) and queries (reading) into different models.\n\n``didiator`` have segregated ``.send(command)``, ``.query(query)`` and ``.publish(events)`` methods in its ``Mediator`` and\nassumes that you will separate its handlers.\nUse ``CommandMediator``, ``QueryMediator`` and ``EventMediator`` protocols to explicitly define which method you need in ``YourController``\n\n.. code-block:: mermaid\n\n graph LR;\n YourController-- Query -->Mediator;\n YourController-- Command -->Mediator;\n Mediator-. Query .->QueryDispatcher-.->di2[DiMiddleware]-.->QueryHandler;\n Mediator-. Command .->CommandDispatcher-.->di1[DiMiddleware]-.->CommandHandler;\n CommandHandler-- Event -->Mediator;\n Mediator-. Event .->EventObserver-.->di3[DiMiddleware]-.->EventHandler1;\n EventObserver-.->di4[DiMiddleware]-.->EventHandler2;\n\n``DiMiddleware`` initializes handlers and injects dependencies for them, you can just send a command with the data you need\n\nWhy ``didiator``?\n=================\n\n- Easy dependency injection to your business logic\n- Separating dependencies from your controllers. They can just parse external requests and interact with the ``Mediator``\n- CQRS\n- Event publishing\n- Flexible configuration\n- Middlewares support\n\nWhy not?\n========\n\n- You don't need it\n- Maybe too low coupling: navigation becomes more difficult\n- Didiator is in beta now\n- No support for synchronous handlers\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A library that implements the Mediator pattern and uses DI library",
"version": "0.3.0",
"split_keywords": [
"didiator",
"mediatr",
"mediator",
"cqrs",
"di",
"events",
"ioc"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "593a5787e2ec6ad831d0ac80ffc0f97b3c73b627dcd2c62455d4a5deb992a634",
"md5": "2cad907ec61b12dd9fa0edd20cc98414",
"sha256": "6cf5af01023e0e5cac45460e02769bb6ad64e1332df346aaefb46686a29297b0"
},
"downloads": -1,
"filename": "didiator-0.3.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2cad907ec61b12dd9fa0edd20cc98414",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10,<4.0",
"size": 22056,
"upload_time": "2023-01-30T23:15:34",
"upload_time_iso_8601": "2023-01-30T23:15:34.811532Z",
"url": "https://files.pythonhosted.org/packages/59/3a/5787e2ec6ad831d0ac80ffc0f97b3c73b627dcd2c62455d4a5deb992a634/didiator-0.3.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "20d78f577ae09ed6ee6b5df6cb669e34d7cc848c80a7a2045b3e9f1921c80ded",
"md5": "95f3d798e29eecd82b65dd13fb33d89d",
"sha256": "69cdbdfdabbe38d02bfbf93abd739098e0bab0f83bf8bc92973800fdc436b78d"
},
"downloads": -1,
"filename": "didiator-0.3.0.tar.gz",
"has_sig": false,
"md5_digest": "95f3d798e29eecd82b65dd13fb33d89d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10,<4.0",
"size": 14207,
"upload_time": "2023-01-30T23:15:36",
"upload_time_iso_8601": "2023-01-30T23:15:36.810661Z",
"url": "https://files.pythonhosted.org/packages/20/d7/8f577ae09ed6ee6b5df6cb669e34d7cc848c80a7a2045b3e9f1921c80ded/didiator-0.3.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-01-30 23:15:36",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "SamWarden",
"github_project": "didiator",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "didiator"
}
JSON
