使用 Python 进行 DDD：我做对了吗？ [关闭]

Question

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

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

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

下面代码的相关部分。请仔细查看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)
        # ...

我可以做些什么来获得一个没有开销的干净架构？

Answer 1

为了回答您的问题，我开始了一篇博文，您可以在这里找到：https://lukeonpython.blog/2020/04/my-structure-for-ddd-component/。 目前你只有代码sn-ps，稍后我会添加一些描述:-)。

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

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

所以这里有一个用于组件包结构的代码 sn-ps 的小示例。我使用这样的东西：

• 迁移/
• app.py
• commands.py
• events.py
• exceptions.py
• repository.py
• service.py
• uow.py

在迁移中，我更喜欢为这个特定组件使用带有分支的 alembic。所以不会对项目中的其他组件产生依赖。

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

对于其余的模块，我将在这里给出一些sn-ps。

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/

你可以在这里找到这个例子的完整来源https://github.com/lzukowski/lzukowski.github.io/tree/master/examples/ddd_component