I'm trying to use Domain Driven Design (DDD) in a Python project, but it looks like a lot of boilerplate code. I think I'm on the wrong path.
I've three files, all defining the item for each purpose. It feels too much. Also I'm converting to and from dictionaries too often, but I'm trying to keep the purposes separated.
This topic should not be opinion-based, because I'm trying to follow the DDD approach and there should be a pattern to follow.
Relevant part of the code below. Please have a closer look on the ItemRepository.
/domain/item.py
"""
Vanilla Python class, business level
"""
class ItemDomain:
def __init__(self, name):
self.name = name
@classmethod
def from_dictionary(cls, dictionary):
return cls(name=dictionary['name'])
def to_dictionary(self):
return {'name': self.name }
/model/item.py
"""
Persistent model for SQLAlchemy
"""
class ItemModel(DefaultModel):
__tablename__ = 'items'
name = Column(Text)
/schema/item.py
"""
Schema for Marshmallow
"""
class ItemSchema(Schema):
name = fields.Str(required=True)
/repository/item.py
class ItemRepository:
def get_one(item_id):
# ...
model = session.query(ItemModel).filter_by(item_id=item_id).first()
return ItemDomain.from_dictionary(dict(model))
def add_one(item: ItemDomain):
# ...
item = item.to_dictionary()
ItemSchema().load(item) # validation: will raise an exception if invalid
model = ItemModel()
model.from_dictionary(item)
session.add(model)
# ...
What can I do to have a clean architecture without overhead?
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:
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