使用 Python 进行 DDD：我做对了吗？

Question

我正在尝试在 Python 项目中使用领域驱动设计 (DDD)，但它看起来像很多样板代码。我想我走错了路。

我有三个文件，都为每个目的定义了项目。感觉太多了。此外，我经常在字典之间进行转换，但我试图将目的分开。

这个话题不应该是基于意见的，因为我正在尝试遵循 DDD 方法并且应该有一个模式可以遵循。

下面代码的相关部分。请仔细看看ItemRepository。

/域/项目.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 } 

/模型/项目.py

"""
Persistent model for SQLAlchemy
"""
class ItemModel(DefaultModel):
    __tablename__ = 'items'
    name = Column(Text)

/架构/项目.py

"""
Schema for Marshmallow
"""
class ItemSchema(Schema):
    name = fields.Str(required=True)

/存储库/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)
        # ...

我该怎么做才能拥有一个干净的架构而没有开销？

Answer 1

为了回答你的问题，我开始写一篇博文，你可以在这里找到：https : //lzukowski.github.io/2020-04-14/ddd_component_in_python/。目前你只有代码片段，稍后我会添加一些描述:-)。

但一般来说，DDD 应该是一个独立的组件，具有由纯数据对象完成的通信的外观。这个门面是一个应用服务。就我而言，它是命令处理程序和查询处理程序。大多数测试是使用 Facade 的 BDD 测试。有时对于复杂的域逻辑，您可以在聚合/UnitOfWork 上使用单元测试。您的应用程序架构将 DDD 元素拆分为我不喜欢的不同包。使用这种方法，您将失去对组件边界的控制。您需要从该组件中获得的所有内容都应导出到init .py。通常，它是一个带有命令的命令处理程序。如果您需要数据视图，请使用查询处理程序。事件侦听器注册可能的事件。

如果您不确定是否需要所有这些东西，您可以从外观上的 BDD 测试和内部非常简化的实现开始。因此，具有直接使用 DTO 的业务逻辑的命令处理程序。稍后如果事情变得复杂，您可以轻松重构。但适当的界限是成功的关键。另外，请记住，如果您觉得所有这些元素和代码都是开销，那么您可能不需要 DDD 方法。也许它不符合 DDD 的条件。

因此，这里有一个包含组件包结构代码片段的小示例。我使用这样的东西：-迁移/-app.py-commands.py-events.py-exceptions.py-repository.py-service.py-uow.py

在迁移中，我更喜欢将 alembic 与这个特定组件的分支一起使用。因此不会对项目中的其他组件产生依赖。

app.py 是具有依赖注入的容器的地方。它主要用于将适当的存储库注入应用程序服务和存储库依赖项。

对于其余的模块，我将在这里给出一些片段。

命令.py

@dataclass
class Create(Command):
    command_id: CommandID = field(default_factory=uuid1)
    timestamp: datetime = field(default_factory=datetime.utcnow

服务.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)

文件

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()

存储库.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()

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()

https://lzukowski.github.io/2020-04-14/ddd_component_in_python/

您可以在此处找到此示例的完整来源https://github.com/lzukowski/lzukowski.github.io/tree/master/examples/ddd_component