Skip to main content

Async Python DDD utilities library.

Project description

Async Python DDD utilities library

PyPI version PyPIDownloads CI

aioddd is an async Python DDD utilities library.

Installation

Use the package manager pip to install aioddd.

pip install aioddd

Documentation

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.9

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

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aioddd-1.5.0.tar.gz (13.3 kB view details)

Uploaded Source

Built Distribution

aioddd-1.5.0-py3-none-any.whl (12.3 kB view details)

Uploaded Python 3

File details

Details for the file aioddd-1.5.0.tar.gz.

File metadata

  • Download URL: aioddd-1.5.0.tar.gz
  • Upload date:
  • Size: 13.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.13.0rc2

File hashes

Hashes for aioddd-1.5.0.tar.gz
Algorithm Hash digest
SHA256 1cc96309aab329a6a62f2f8d7058ec0872d0fd882d6b5ee1c220687356bf8545
MD5 965cad5ece748a669e48316e1bdd9c47
BLAKE2b-256 4d7e0e59e220ee9f4c9b7bba822790f08c7ebb991e9e45c1677ecae3a263ba8d

See more details on using hashes here.

File details

Details for the file aioddd-1.5.0-py3-none-any.whl.

File metadata

  • Download URL: aioddd-1.5.0-py3-none-any.whl
  • Upload date:
  • Size: 12.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.13.0rc2

File hashes

Hashes for aioddd-1.5.0-py3-none-any.whl
Algorithm Hash digest
SHA256 c7d0fbf9d4f678cb43b63a72f95eb8cd442e0b8312eb3d5ea754d8ba44ff3911
MD5 a8e82c3f52f00f7e2c2ef1f422da66e2
BLAKE2b-256 7384c3f56765f0a9d08c922a0a9bbe0b57621f4f775a6bc48bbd173acfdf7d18

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page