To answer your question I started a blog post that you can find here: https://lukeonpython.blog/2020/04/my-structure-for-ddd-component/ .
At this moment you have only code snippets, later I will add some description :-).
But generally, DDD should be an independent component with a facade for communication done by pure data objects. This facade is an application service. In my case, it's command handlers and query handlers. Most tests are BDD tests using facade. Sometimes with complicated domain logic, you can use unit test on aggregate/UnitOfWork.
Your app architecture splits DDD elements into different packages which I don't like. With this approach, you lose control over component boundaries. All things that you need from this component should be exported to init.py.
Generally, it's a command handler with command. Query handler if you have a need for the data view. Event listener registration with possible events.
If you are not sure if you need all these things you can start with BDD tests on facade and very simplified implementation inside. So Command Handler with business logic that is using DTO directly. Later if things will complicate you can refactor easily. But proper boundaries is key for success. Also, remember that maybe you don't need DDD approach if you feel that all this elements and code are overhead. Maybe it's not qualifying for DDD.
So here is a little example with code snippets for a component package structure. I use something like this:
- migrations/
- app.py
- commands.py
- events.py
- exceptions.py
- repository.py
- service.py
- uow.py
In migrations, I prefer to use alembic with branches for this specific component. So there will be no dependency on other components in the project.
app.py is a place for container with dependency injection. It's mostly for injecting proper repository to application service and repository dependencies.
For the rest of modules, I will give some snippets here.
commands.py
@dataclass
class Create(Command):
command_id: CommandID = field(default_factory=uuid1)
timestamp: datetime = field(default_factory=datetime.utcnow
service.py
class CommandHandler:
def __init__(self, repository: Repository) -> None:
self._repository = repository
self._listeners: List[Listener] = []
super().__init__()
def register(self, listener: Listener) -> None:
if listener not in self._listeners:
self._listeners.append(listener)
def unregister(self, listener: Listener) -> None:
if listener in self._listeners:
self._listeners.remove(listener)
@safe
@singledispatchmethod
def handle(self, command: Command) -> Optional[Event]:
uow: UnitOfWork = self._repository.get(command.uow_id)
event: Event = app_event(self._handle(command, uow), command)
for listener in self._listeners:
listener(event)
self._repository.save(uow)
return event
@safe
@handle.register(Create)
def create(self, command: Create) -> Event:
uow = UnitOfWork.create()
self._repository.save(uow)
return Created(command.command_id, uow.id)
@singledispatchmethod
def _handle(self, c: Command, u: UnitOfWork) -> UnitOfWork.Event:
raise NotImplementedError
@_handle.register(UpdateValue)
def _(self, command: UpdateValue, uow: UnitOfWork) -> UnitOfWork.Event:
return uow.update(command.value)
uow.py
UnitOfWorkID = NewType('UnitOfWorkID', UUID)
class UnitOfWorkDTO:
id: UnitOfWorkID
value: Optional[Text]
class UnitOfWork:
id: UnitOfWorkID
dto: UnitOfWorkDTO
class Event:
pass
class Updated(Event):
pass
def __init__(self, dto: UnitOfWorkDTO) -> None:
self.id = dto.id
self.dto = dto
@classmethod
def create(cls) -> 'UnitOfWork':
dto = UnitOfWorkDTO()
dto.id = UnitOfWorkID(uuid1())
dto.value = None
return UnitOfWork(dto)
def update(self, value: Text) -> Updated:
self.dto.value = value
return self.Updated()
repository.py
class ORMRepository(Repository):
def __init__(self, session: Session):
self._session = session
self._query = self._session.query(UnitOfWorkMapper)
def get(self, uow_id: UnitOfWorkID) -> UnitOfWork:
dto = self._query.filter_by(uuid=uow_id).one_or_none()
if not dto:
raise NotFound(uow_id)
return UnitOfWork(dto)
def save(self, uow: UnitOfWork) -> None:
self._session.add(uow.dto)
self._session.flush()
entities_t = Table = Table(
'entities',
meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('uuid', String, unique=True, index=True),
Column('value', String, nullable=True),
)
UnitOfWorkMapper = mapper(
UnitOfWorkDTO,
entities_t,
properties={
'id': entities_t.c.uuid,
'value': entities_t.c.value,
},
column_prefix='_db_column_',
)
https://lukeonpython.blog/2020/04/my-structure-for-ddd-component/
Full sources of this example you can find here https://github.com/lzukowski/lzukowski.github.io/tree/master/examples/ddd_component
