python - 使用 Python 进行 DDD:我做对了吗?
问题描述
我正在尝试在 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)
# ...
我可以做些什么来拥有一个没有开销的干净架构?
解决方案
为了回答你的问题,我开始了一篇博文,你可以在这里找到:https ://lukeonpython.blog/2020/04/my-structure-for-ddd-component/ 。目前你只有代码片段,稍后我会添加一些描述:-)。
但一般来说,DDD 应该是一个独立的组件,具有由纯数据对象完成的通信的外观。这个门面是一个应用服务。就我而言,它是命令处理程序和查询处理程序。大多数测试是使用外观的 BDD 测试。有时对于复杂的域逻辑,您可以在聚合/UnitOfWork 上使用单元测试。您的应用架构将 DDD 元素拆分为我不喜欢的不同包。使用这种方法,您将失去对组件边界的控制。你需要从这个组件中得到的所有东西都应该导出到init .py。通常,它是一个带有命令的命令处理程序。如果您需要数据视图,请查询处理程序。带有可能事件的事件侦听器注册。
如果您不确定是否需要所有这些东西,您可以从外观上的 BDD 测试和非常简化的内部实现开始。因此,具有直接使用 DTO 的业务逻辑的命令处理程序。以后如果事情变得复杂,你可以轻松地重构。但适当的界限是成功的关键。另外,请记住,如果您觉得所有这些元素和代码都是开销,那么您可能不需要 DDD 方法。也许它不符合 DDD 的条件。
因此,这里有一个包含组件包结构的代码片段的小示例。我使用这样的东西:
- 迁移/
- 应用程序.py
- 命令.py
- 事件.py
- 异常.py
- 存储库.py
- 服务.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)
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()
存储库.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
推荐阅读
- java - 设置默认回滚规则时,事务回滚不适用于 Propagation.REQUIRES_NEW
- api - 如何使用不同的输入查询参数发送不同的 RAML 响应
- php - 致命错误:未捕获的错误:使用请求时未找到类“Requests_Hooks”
- python - roc_curve pos_label 混淆
- amazon-web-services - AWS - 同一域的跨区域 SSL 证书
- vba - 将 lastrow 变量设置为从最后占用的行偏移 3 行
- node.js - aws-sdk 未将映像部署到 s3 存储桶
- c# - 如何关闭由 spire.xls 创建的文件
- sql - 任何人都可以从我拿起的存储过程中解释这个片段吗
- javascript - TypeError:Functionname 不是 HTMLButtonElement.onClick 处的函数