Async Python DDD utilities library
aioddd is an async Python DDD utilities library.
Installation
Use the package manager pip to install aioddd.
pip install aioddd
Documentation
- Visit aioddd docs.
Usage
from asyncio import get_event_loop
from dataclasses import dataclass
from typing import Type
from aioddd import NotFoundError, \
Command, CommandHandler, SimpleCommandBus, \
Query, QueryHandler, OptionalResponse, SimpleQueryBus, Event
_products = []
class ProductStored(Event):
@dataclass
class Attributes:
ref: str
attributes: Attributes
class StoreProductCommand(Command):
def __init__(self, ref: str):
self.ref = ref
class StoreProductCommandHandler(CommandHandler):
def subscribed_to(self) -> Type[Command]:
return StoreProductCommand
async def handle(self, command: StoreProductCommand) -> None:
_products.append(command.ref)
class ProductNotFoundError(NotFoundError):
_code = 'product_not_found'
_title = 'Product not found'
class FindProductQuery(Query):
def __init__(self, ref: str):
self.ref = ref
class FindProductQueryHandler(QueryHandler):
def subscribed_to(self) -> Type[Query]:
return FindProductQuery
async def handle(self, query: FindProductQuery) -> OptionalResponse:
if query.ref != '123':
raise ProductNotFoundError.create(detail={'ref': query.ref})
return {'ref': query.ref}
async def main() -> None:
commands_bus = SimpleCommandBus([StoreProductCommandHandler()])
await commands_bus.dispatch(StoreProductCommand('123'))
query_bus = SimpleQueryBus([FindProductQueryHandler()])
response = await query_bus.ask(FindProductQuery('123'))
print(response)
if __name__ == '__main__':
get_event_loop().run_until_complete(main())
Requirements
- Python >= 3.10
Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
Please make sure to update tests as appropriate.
License
1 maintainer
Metadata
1 maintainer
Release files for aioddd 1.6.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| aioddd-1.6.0.tar.gz | 13.4 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| aioddd-1.6.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 25.8 kB
Release files / aioddd-1.6.0.tar.gz
| Download URL | aioddd-1.6.0.tar.gz |
|---|---|
| Size | 13.4 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
0c679e525028352e72f923c0f001ef7dfcda172ace22c9d91bc477960af19e38
|
|
BLAKE2b-256 checksum How to use checksums |
9cea5d0b804fd095c0eb5bc46b61db1a8557ee5a55630245601a45de6dff6fd2
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.13.5
|
Release files / aioddd-1.6.0-py3-none-any.whl
| Download URL | aioddd-1.6.0-py3-none-any.whl |
|---|---|
| Size | 12.4 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
b940b18e7e478bd0218fad87f31be6c53df5f83d177344dee3ce9b75aaf9b84e
|
|
BLAKE2b-256 checksum How to use checksums |
172937365d07f40314b3ea2d923f89bf7eef966caf2be96b4f060269b200597a
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.13.5
|
