<a href="https://github.com/akhundMurad/diator/actions?query=setup%3ACI%2FCD+event%3Apush+branch%3Amain" target="_blank">
<img src="https://github.com/akhundMurad/diator/actions/workflows/setup.yml/badge.svg?event=push&branch=main" alt="Test">
</a>
<a href="https://pepy.tech/project/diator" target="_blank">
<img src="https://static.pepy.tech/personalized-badge/diator?period=total&units=international_system&left_color=black&right_color=red&left_text=downloads" alt="Downloads">
</a>
<a href="https://pypi.org/project/diator" target="_blank">
<img src="https://img.shields.io/pypi/v/diator?color=red&labelColor=black" alt="Package version">
</a>
<a href="https://pypi.org/project/diator" target="_blank">
<img src="https://img.shields.io/pypi/pyversions/diator.svg?color=red&labelColor=black" alt="Supported Python versions">
</a>
# Diator - CQRS Library for Python
**[Docs](https://akhundmurad.github.io/diator/) | [PyPI](https://pypi.org/project/diator/)**
Diator is a Python library for implementing CQRS pattern in your Python applications. It provides a set of abstractions and utilities to help you separate your read and write concerns, allowing for better scalability, performance, and maintainability of your application.
## Features :bulb:
- Implements the CQRS pattern.
- Simple, yet flexible API.
- Supports multiple message brokers, such as [Redis Pub/Sub](https://redis.io/docs/manual/pubsub/) and [Azure Service Bus](https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview).
- Supports various di-frameworks, such as [di](https://github.com/adriangb/di) and [rodi](https://github.com/Neoteroi/rodi).
- Easy to integrate with existing codebases.
## Installation :triangular_ruler:
Install the Diator library with [pip](https://pypi.org/project/diator/)
```bash
pip install diator
```
There are also several installation options:
- To use Redis as Message Broker
```bash
pip install diator[redis]
```
- Or Azure Service Bus
```bash
pip install diator[azure]
```
## Simple Example :hammer_and_wrench:
Minimal example of diator usage:
```python
import asyncio
from dataclasses import dataclass, field
from di import Container, bind_by_type
from di.dependent import Dependent
from diator.events import EventMap, Event, EventEmitter
from diator.container.di import DIContainer
from diator.mediator import Mediator
from diator.requests import Request, RequestHandler, RequestMap
@dataclass(frozen=True, kw_only=True)
class JoinMeetingCommand(Request):
meeting_id: int
user_id: int
is_late: bool = field(default=False)
class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
def __init__(self, meeting_api) -> None:
self._meeting_api = meeting_api
self._events: list[Event] = []
@property
def events(self) -> list[Event]:
return self._events
async def handle(self, request: JoinMeetingCommand) -> None:
self._meeting_api.join(request.meeting_id, request.user_id)
if request.is_late:
self._meeting_api.warn(request.user_id)
def setup_di() -> DIContainer:
external_container = Container()
external_container.bind(
bind_by_type(
Dependent(JoinMeetingCommandHandler, scope="request"),
JoinMeetingCommandHandler,
)
)
container = DIContainer()
container.attach_external_container(external_container)
return container
async def main() -> None:
container = setup_di()
request_map = RequestMap()
request_map.bind(JoinMeetingCommand, JoinMeetingCommandHandler)
event_emitter = EventEmitter(
event_map=EventMap(), container=container, message_broker=None
)
mediator = Mediator(
request_map=request_map,
event_emitter=event_emitter,
container=container,
)
await mediator.send(JoinMeetingCommand(user_id=1, meeting_id=1, is_late=True))
if __name__ == "__main__":
asyncio.run(main())
```
## Further reading :scroll:
- [Udi Dahan - Clarified CQRS](https://udidahan.com/2009/12/09/clarified-cqrs/)
- [Martin Fowler - CQRS](https://martinfowler.com/bliki/CQRS.html)
- [Marting Fowler - What do you mean by “Event-Driven”?](https://martinfowler.com/articles/201701-event-driven.html)
- [Vlad Khononov - Learning Domain-Driven Design](https://www.oreilly.com/library/view/learning-domain-driven-design/9781098100124/)
- [Vaughn Vernon - Really Simple CQRS](https://kalele.io/really-simple-cqrs/)
## License
This project is licensed under the terms of the MIT license.
Raw data
{
"_id": null,
"home_page": "",
"name": "diator",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "",
"keywords": "CQRS,async,asyncio,command,commands,di,diator,event,events,mediator,mediatr,queries,query",
"author": "",
"author_email": "Murad Akhundov <akhundov1murad@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/7f/43/b908d8bc852fa275e80e54d9883b35f4f6326d36b815e006f9edce394388/diator-0.1.2.tar.gz",
"platform": null,
"description": "\n\n<a href=\"https://github.com/akhundMurad/diator/actions?query=setup%3ACI%2FCD+event%3Apush+branch%3Amain\" target=\"_blank\">\n <img src=\"https://github.com/akhundMurad/diator/actions/workflows/setup.yml/badge.svg?event=push&branch=main\" alt=\"Test\">\n</a>\n<a href=\"https://pepy.tech/project/diator\" target=\"_blank\">\n <img src=\"https://static.pepy.tech/personalized-badge/diator?period=total&units=international_system&left_color=black&right_color=red&left_text=downloads\" alt=\"Downloads\">\n</a>\n<a href=\"https://pypi.org/project/diator\" target=\"_blank\">\n <img src=\"https://img.shields.io/pypi/v/diator?color=red&labelColor=black\" alt=\"Package version\">\n</a>\n<a href=\"https://pypi.org/project/diator\" target=\"_blank\">\n <img src=\"https://img.shields.io/pypi/pyversions/diator.svg?color=red&labelColor=black\" alt=\"Supported Python versions\">\n</a>\n\n# Diator - CQRS Library for Python\n**[Docs](https://akhundmurad.github.io/diator/) | [PyPI](https://pypi.org/project/diator/)**\n\nDiator is a Python library for implementing CQRS pattern in your Python applications. It provides a set of abstractions and utilities to help you separate your read and write concerns, allowing for better scalability, performance, and maintainability of your application.\n\n## Features :bulb:\n\n- Implements the CQRS pattern.\n- Simple, yet flexible API.\n- Supports multiple message brokers, such as [Redis Pub/Sub](https://redis.io/docs/manual/pubsub/) and [Azure Service Bus](https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview).\n- Supports various di-frameworks, such as [di](https://github.com/adriangb/di) and [rodi](https://github.com/Neoteroi/rodi).\n- Easy to integrate with existing codebases.\n\n## Installation :triangular_ruler:\n\nInstall the Diator library with [pip](https://pypi.org/project/diator/)\n\n```bash\npip install diator\n```\n\nThere are also several installation options:\n\n- To use Redis as Message Broker\n\n ```bash\n pip install diator[redis]\n ```\n\n- Or Azure Service Bus\n\n ```bash\n pip install diator[azure]\n ```\n\n## Simple Example :hammer_and_wrench:\n\nMinimal example of diator usage:\n\n```python\nimport asyncio\nfrom dataclasses import dataclass, field\nfrom di import Container, bind_by_type\nfrom di.dependent import Dependent\nfrom diator.events import EventMap, Event, EventEmitter\nfrom diator.container.di import DIContainer\nfrom diator.mediator import Mediator\nfrom diator.requests import Request, RequestHandler, RequestMap\n\n\n@dataclass(frozen=True, kw_only=True)\nclass JoinMeetingCommand(Request):\n meeting_id: int\n user_id: int\n is_late: bool = field(default=False)\n\n\nclass JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):\n def __init__(self, meeting_api) -> None:\n self._meeting_api = meeting_api\n self._events: list[Event] = []\n\n @property\n def events(self) -> list[Event]:\n return self._events\n\n async def handle(self, request: JoinMeetingCommand) -> None:\n self._meeting_api.join(request.meeting_id, request.user_id)\n if request.is_late:\n self._meeting_api.warn(request.user_id)\n\n\ndef setup_di() -> DIContainer:\n external_container = Container()\n\n external_container.bind(\n bind_by_type(\n Dependent(JoinMeetingCommandHandler, scope=\"request\"),\n JoinMeetingCommandHandler,\n )\n )\n\n container = DIContainer()\n container.attach_external_container(external_container)\n\n return container\n\n\nasync def main() -> None:\n container = setup_di()\n\n request_map = RequestMap()\n request_map.bind(JoinMeetingCommand, JoinMeetingCommandHandler)\n\n event_emitter = EventEmitter(\n event_map=EventMap(), container=container, message_broker=None\n )\n\n mediator = Mediator(\n request_map=request_map,\n event_emitter=event_emitter,\n container=container,\n )\n\n await mediator.send(JoinMeetingCommand(user_id=1, meeting_id=1, is_late=True))\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n\n```\n\n## Further reading :scroll:\n\n- [Udi Dahan - Clarified CQRS](https://udidahan.com/2009/12/09/clarified-cqrs/)\n- [Martin Fowler - CQRS](https://martinfowler.com/bliki/CQRS.html)\n- [Marting Fowler - What do you mean by \u201cEvent-Driven\u201d?](https://martinfowler.com/articles/201701-event-driven.html)\n- [Vlad Khononov - Learning Domain-Driven Design](https://www.oreilly.com/library/view/learning-domain-driven-design/9781098100124/)\n- [Vaughn Vernon - Really Simple CQRS](https://kalele.io/really-simple-cqrs/)\n\n## License\n\nThis project is licensed under the terms of the MIT license.\n",
"bugtrack_url": null,
"license": "",
"summary": "Diator is a Python library for implementing CQRS pattern in your Python applications.",
"version": "0.1.2",
"project_urls": {
"Bug Tracker": "https://github.com/akhundMurad/diator/issues",
"Homepage": "https://github.com/akhundMurad/diator"
},
"split_keywords": [
"cqrs",
"async",
"asyncio",
"command",
"commands",
"di",
"diator",
"event",
"events",
"mediator",
"mediatr",
"queries",
"query"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0723906766e984b53d3abe63c9ae2a2c6990269dcaa0a7be106c39226657b321",
"md5": "7ed3af4a27cf63f5c02be3432169fb07",
"sha256": "2a252c4c2cb9963b9fb24dbc4e8c7b63f3bf880501505b07ef7bffb356ac3c09"
},
"downloads": -1,
"filename": "diator-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7ed3af4a27cf63f5c02be3432169fb07",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 16544,
"upload_time": "2023-05-02T22:03:32",
"upload_time_iso_8601": "2023-05-02T22:03:32.258442Z",
"url": "https://files.pythonhosted.org/packages/07/23/906766e984b53d3abe63c9ae2a2c6990269dcaa0a7be106c39226657b321/diator-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "7f43b908d8bc852fa275e80e54d9883b35f4f6326d36b815e006f9edce394388",
"md5": "60f0cb1ec5bf0e299de1465a6e833212",
"sha256": "da44cfec719fd7dc0b33eda7295ae31ad0afb379489778addb5c85eea0755d8f"
},
"downloads": -1,
"filename": "diator-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "60f0cb1ec5bf0e299de1465a6e833212",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 18143,
"upload_time": "2023-05-02T22:03:35",
"upload_time_iso_8601": "2023-05-02T22:03:35.432064Z",
"url": "https://files.pythonhosted.org/packages/7f/43/b908d8bc852fa275e80e54d9883b35f4f6326d36b815e006f9edce394388/diator-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-05-02 22:03:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "akhundMurad",
"github_project": "diator",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "diator"
}
JSON
