aioddd is an async Python DDD utilities library.
Use the package manager pip to install aioddd.
pip install aioddd- Visit aioddd docs.
fromasyncioimportget_event_loopfromdataclassesimportdataclassfromtypingimportTypefromaiodddimportNotFoundError, \
Command, CommandHandler, SimpleCommandBus, \
Query, QueryHandler, OptionalResponse, SimpleQueryBus, Event_products= []
classProductStored(Event):
@dataclassclassAttributes:
ref: strattributes: AttributesclassStoreProductCommand(Command):
def__init__(self, ref: str):
self.ref=refclassStoreProductCommandHandler(CommandHandler):
defsubscribed_to(self) ->Type[Command]:
returnStoreProductCommandasyncdefhandle(self, command: StoreProductCommand) ->None:
_products.append(command.ref)
classProductNotFoundError(NotFoundError):
_code='product_not_found'_title='Product not found'classFindProductQuery(Query):
def__init__(self, ref: str):
self.ref=refclassFindProductQueryHandler(QueryHandler):
defsubscribed_to(self) ->Type[Query]:
returnFindProductQueryasyncdefhandle(self, query: FindProductQuery) ->OptionalResponse:
ifquery.ref!='123':
raiseProductNotFoundError.create(detail={'ref': query.ref})
return {'ref': query.ref}
asyncdefmain() ->None:
commands_bus=SimpleCommandBus([StoreProductCommandHandler()])
awaitcommands_bus.dispatch(StoreProductCommand('123'))
query_bus=SimpleQueryBus([FindProductQueryHandler()])
response=awaitquery_bus.ask(FindProductQuery('123'))
print(response)
if__name__=='__main__':
get_event_loop().run_until_complete(main())- Python >= 3.10
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.