2：仓库模式
现在是时候兑现我们的承诺，使用依赖倒置原则来解耦我们的核心逻辑与基础设施关注点了。
我们将介绍仓库模式，这是一种对数据存储的简化抽象，使我们能够将模型层与数据层解耦。我们将展示一个具体的例子，说明这种简化抽象如何通过隐藏数据库的复杂性来提高系统的可测试性。
仓库模式之前和之后 展示了我们将要构建内容的一个小预览：一个位于我们的领域模型和数据库之间的 Repository 对象。
|
提示
|
本章的代码位于 chapter_02_repository 分支 GitHub 上。 git clone https://github.com/cosmicpython/code.git cd code git checkout chapter_02_repository # or to code along, checkout the previous chapter: git checkout chapter_01_domain_model |
持久化我们的领域模型
在 [chapter_01_domain_model] 中，我们构建了一个简单的领域模型，可以将订单分配给批次库存。我们很容易针对此代码编写测试，因为没有任何依赖项或基础设施需要设置。如果我们需要运行数据库或 API 并创建测试数据，我们的测试将更难编写和维护。
遗憾的是，在某些时候，我们需要将我们完美的小模型交付给用户，并应对电子表格、Web 浏览器和竞争条件的真实世界。在接下来的几章中，我们将研究如何将我们理想化的领域模型连接到外部状态。
我们希望以敏捷的方式工作，因此我们的首要任务是尽快获得最小可行产品。在我们的例子中，那将是一个 Web API。在真实的项目中，您可能会直接进行一些端到端测试，并开始插入 Web 框架，从外部开始进行测试驱动。
但我们知道，无论如何，我们都需要某种形式的持久存储，而且这是一本教科书，因此我们可以稍微允许自己进行更多的自下而上的开发，并开始考虑存储和数据库。
一些伪代码：我们需要什么？
当我们构建我们的第一个 API 端点时，我们知道我们将有一些代码或多或少如下所示。
@flask.route.gubbins
def allocate_endpoint():
# extract order line from request
line = OrderLine(request.params, ...)
# load all batches from the DB
batches = ...
# call our domain service
allocate(line, batches)
# then save the allocation back to the database somehow
return 201|
注意
|
我们使用了 Flask，因为它很轻巧，但是您无需成为 Flask 用户即可理解本书。实际上，我们将向您展示如何使您对框架的选择成为次要细节。 |
我们将需要一种方法从数据库中检索批次信息并从中实例化我们的领域模型对象，并且我们还需要一种方法将它们保存回数据库。
什么？哦，“gubbins”是英国英语中“东西”的意思。您可以忽略它。这是伪代码，好吗？
将 DIP 应用于数据访问
Django 的 Model-View-Template 结构与 Model-View-Controller (MVC) 密切相关。在任何情况下，目的都是保持层之间的分离（这是一件好事），并使每一层仅依赖于其下的一层。
但是我们希望我们的领域模型没有任何依赖项。[1] 我们不希望基础设施问题渗透到我们的领域模型中，从而减慢我们的单元测试或我们进行更改的能力。
相反，正如引言中所讨论的，我们将把我们的模型视为“内部”，并且依赖关系向内流入；这就是人们有时称之为洋葱架构的东西（请参阅 洋葱架构）。
[ditaa, apwp_0203]
+------------------------+
| Presentation Layer |
+------------------------+
|
V
+--------------------------------------------------+
| Domain Model |
+--------------------------------------------------+
^
|
+---------------------+
| Database Layer |
+---------------------+
提醒：我们的模型
让我们提醒自己我们的领域模型（请参阅 我们的模型）：分配是将 OrderLine 链接到 Batch 的概念。我们将分配存储为 Batch 对象上的集合。
让我们看看如何将其转换为关系数据库。
“正常”的 ORM 方式：模型依赖于 ORM
如今，您的团队成员不太可能手动编写自己的 SQL 查询。相反，您几乎肯定会使用某种框架来根据您的模型对象为您生成 SQL。
这些框架称为对象关系映射器 (ORM)，因为它们的存在是为了弥合对象和领域建模世界与数据库和关系代数世界之间的概念差距。
ORM 给我们的最重要的事情是持久性无知：我们的花哨领域模型不需要知道数据是如何加载或持久化的。这有助于保持我们的领域免受对特定数据库技术的直接依赖。[3]
但是，如果您遵循典型的 SQLAlchemy 教程，您最终会得到类似这样的东西
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class Order(Base):
id = Column(Integer, primary_key=True)
class OrderLine(Base):
id = Column(Integer, primary_key=True)
sku = Column(String(250))
qty = Integer(String(250))
order_id = Column(Integer, ForeignKey('order.id'))
order = relationship(Order)
class Allocation(Base):
...您无需了解 SQLAlchemy 即可看到我们原始的模型现在充满了对 ORM 的依赖，并且开始看起来非常丑陋。我们真的可以说这个模型对数据库一无所知吗？当我们的模型属性直接耦合到数据库列时，它怎么能与存储关注点分离呢？
倒置依赖关系：ORM 依赖于模型
好吧，谢天谢地，这不是使用 SQLAlchemy 的唯一方法。另一种方法是分别定义您的模式，并为如何在模式和我们的领域模型之间进行转换定义显式的映射器，SQLAlchemy 称之为 经典映射
from sqlalchemy.orm import mapper, relationship
import model #(1)
metadata = MetaData()
order_lines = Table( #(2)
"order_lines",
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("sku", String(255)),
Column("qty", Integer, nullable=False),
Column("orderid", String(255)),
)
...
def start_mappers():
lines_mapper = mapper(model.OrderLine, order_lines) #(3)-
ORM 导入（或“依赖于”或“了解”）领域模型，而不是相反。
-
我们使用 SQLAlchemy 的抽象来定义我们的数据库表和列。[4]
-
当我们调用
mapper函数时，SQLAlchemy 会施展其魔力，将我们的领域模型类绑定到我们定义的各种表。
最终结果是，如果我们调用 start_mappers，我们将能够轻松地从数据库加载领域模型实例并将其保存到数据库。但是，如果我们从未调用该函数，则我们的领域模型类将完全不知道数据库的存在。
这为我们提供了 SQLAlchemy 的所有好处，包括使用 alembic 进行迁移的能力，以及使用我们的领域类透明地进行查询的能力，我们将在后面看到。
当您第一次尝试构建 ORM 配置时，编写针对它的测试会很有用，如下例所示
def test_orderline_mapper_can_load_lines(session): #(1)
session.execute(
"INSERT INTO order_lines (orderid, sku, qty) VALUES "
'("order1", "RED-CHAIR", 12),'
'("order1", "RED-TABLE", 13),'
'("order2", "BLUE-LIPSTICK", 14)'
)
expected = [
model.OrderLine("order1", "RED-CHAIR", 12),
model.OrderLine("order1", "RED-TABLE", 13),
model.OrderLine("order2", "BLUE-LIPSTICK", 14),
]
assert session.query(model.OrderLine).all() == expected
def test_orderline_mapper_can_save_lines(session):
new_line = model.OrderLine("order1", "DECORATIVE-WIDGET", 12)
session.add(new_line)
session.commit()
rows = list(session.execute('SELECT orderid, sku, qty FROM "order_lines"'))
assert rows == [("order1", "DECORATIVE-WIDGET", 12)]-
如果您没有使用过 pytest，则需要解释此测试的
session参数。您无需担心 pytest 或其 fixture 的细节，本书的目的在于此，但简短的解释是，您可以将测试的常见依赖项定义为“fixture”，pytest 将通过查看其函数参数将其注入到需要它们的测试中。在本例中，它是一个 SQLAlchemy 数据库会话。
您可能不会长期保留这些测试——正如您很快就会看到的，一旦您采取了倒置 ORM 和领域模型依赖关系的步骤，再采取一小步即可实现另一个名为仓库模式的抽象，这将更容易针对其编写测试，并将在以后的测试中提供一个简单的接口来伪造。
但是我们已经实现了倒置传统依赖关系的目标：领域模型保持“纯净”并且免受基础设施关注的影响。我们可以抛弃 SQLAlchemy 并使用不同的 ORM，或者完全不同的持久性系统，而领域模型无需进行任何更改。
根据您在领域模型中执行的操作，尤其是在您远离 OO 范例的情况下，您可能会发现越来越难以使 ORM 产生您需要的确切行为，并且您可能需要修改您的领域模型。[5] 正如架构决策经常发生的那样，您需要考虑权衡。正如 Python 之禅所说，“实用胜过纯粹！”
但是，此时，我们的 API 端点可能看起来像这样，我们可以使其正常工作
@flask.route.gubbins
def allocate_endpoint():
session = start_session()
# extract order line from request
line = OrderLine(
request.json['orderid'],
request.json['sku'],
request.json['qty'],
)
# load all batches from the DB
batches = session.query(Batch).all()
# call our domain service
allocate(line, batches)
# save the allocation back to the database
session.commit()
return 201介绍仓库模式
仓库模式是对持久存储的抽象。它通过假装我们所有的数据都在内存中来隐藏数据访问的无聊细节。
如果我们的笔记本电脑中有无限的内存，我们将不需要笨拙的数据库。相反，我们可以随时使用我们的对象。那会是什么样子？
import all_my_data
def create_a_batch():
batch = Batch(...)
all_my_data.batches.add(batch)
def modify_a_batch(batch_id, new_quantity):
batch = all_my_data.batches.get(batch_id)
batch.change_initial_quantity(new_quantity)即使我们的对象在内存中，我们也需要将它们放在某个地方，以便我们可以再次找到它们。我们的内存数据将允许我们添加新对象，就像列表或集合一样。由于对象在内存中，我们永远不需要调用 .save() 方法；我们只需获取我们关心的对象并在内存中修改它。
抽象的仓库
最简单的仓库只有两个方法：add() 用于将新项目放入仓库中，get() 用于返回先前添加的项目。[6] 我们严格坚持在我们的领域和服务层中使用这些方法进行数据访问。这种自我强加的简单性阻止了我们将我们的领域模型耦合到数据库。
这是我们的仓库的抽象基类 (ABC) 的外观
class AbstractRepository(abc.ABC):
@abc.abstractmethod #(1)
def add(self, batch: model.Batch):
raise NotImplementedError #(2)
@abc.abstractmethod
def get(self, reference) -> model.Batch:
raise NotImplementedError-
Python 提示：
@abc.abstractmethod是使 ABC 在 Python 中真正“起作用”的唯一方法之一。Python 将拒绝让您实例化一个未实现其父类中定义的所有abstractmethods的类。[7] -
raise NotImplementedError很好，但它既不是必要的也不是充分的。实际上，您的抽象方法可以具有子类可以调用的真实行为，如果您真的想这样做的话。
权衡是什么？
你知道他们说经济学家知道一切事物的价格，却不知道任何事物的价值吗？嗯，程序员知道一切事物的优点，却不知道任何事物的权衡。
每当我们在本书中介绍一种架构模式时，我们都会问：“我们从中得到了什么？它又会花费我们什么？”
通常，至少，我们将引入一个额外的抽象层，尽管我们可能希望它能降低整体复杂性，但它确实会在本地增加复杂性，并且在移动部件的原始数量和持续维护方面有成本。
但是，如果您已经朝着 DDD 和依赖倒置路线前进，那么仓库模式可能是本书中最容易的选择之一。就我们的代码而言，我们实际上只是将 SQLAlchemy 抽象 (session.query(Batch)) 换成我们设计的另一个抽象 (batches_repo.get)。
每次我们添加一个新的想要检索的领域对象时，我们都必须在我们的仓库类中编写几行代码，但是作为回报，我们获得了对我们的存储层的简单抽象，我们可以控制它。仓库模式将使我们能够轻松地对我们存储事物的方式进行根本性的更改（请参阅 [appendix_csvs]），并且正如我们将要看到的，它很容易在单元测试中伪造。
此外，仓库模式在 DDD 世界中非常常见，因此，如果您确实与来自 Java 和 C# 世界的程序员合作，他们很可能会认出它。仓库模式 说明了该模式。
[ditaa, apwp_0205]
+-----------------------------+
| Application Layer |
+-----------------------------+
|^
|| /------------------\
||----------| Domain Model |
|| | Objects |
|| \------------------/
V|
+------------------------------+
| Repository |
+------------------------------+
|
V
+------------------------------+
| Database Layer |
+------------------------------+
与往常一样，我们从测试开始。这可能被归类为集成测试，因为我们正在检查我们的代码（仓库）是否与数据库正确集成；因此，测试倾向于将原始 SQL 与对我们自己代码的调用和断言混合在一起。
|
提示
|
与早期的 ORM 测试不同，这些测试非常适合长期保留在您的代码库中，特别是如果您的领域模型的任何部分意味着对象关系映射是非平凡的。 |
def test_repository_can_save_a_batch(session):
batch = model.Batch("batch1", "RUSTY-SOAPDISH", 100, eta=None)
repo = repository.SqlAlchemyRepository(session)
repo.add(batch) #(1)
session.commit() #(2)
rows = session.execute( #(3)
'SELECT reference, sku, _purchased_quantity, eta FROM "batches"'
)
assert list(rows) == [("batch1", "RUSTY-SOAPDISH", 100, None)]-
repo.add()是此处正在测试的方法。 -
我们将
.commit()保留在仓库之外，并使其成为调用者的责任。这样做有优点和缺点；当我们进入 [chapter_06_uow] 时，我们的一些原因将变得更加清楚。 -
我们使用原始 SQL 来验证是否已保存正确的数据。
下一个测试涉及检索批次和分配，因此它更复杂
def insert_order_line(session):
session.execute( #(1)
"INSERT INTO order_lines (orderid, sku, qty)"
' VALUES ("order1", "GENERIC-SOFA", 12)'
)
[[orderline_id]] = session.execute(
"SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku",
dict(orderid="order1", sku="GENERIC-SOFA"),
)
return orderline_id
def insert_batch(session, batch_id): #(2)
...
def test_repository_can_retrieve_a_batch_with_allocations(session):
orderline_id = insert_order_line(session)
batch1_id = insert_batch(session, "batch1")
insert_batch(session, "batch2")
insert_allocation(session, orderline_id, batch1_id) #(2)
repo = repository.SqlAlchemyRepository(session)
retrieved = repo.get("batch1")
expected = model.Batch("batch1", "GENERIC-SOFA", 100, eta=None)
assert retrieved == expected # Batch.__eq__ only compares reference #(3)
assert retrieved.sku == expected.sku #(4)
assert retrieved._purchased_quantity == expected._purchased_quantity
assert retrieved._allocations == { #(4)
model.OrderLine("order1", "GENERIC-SOFA", 12),
}-
这测试了读取端，因此原始 SQL 正在准备要由
repo.get()读取的数据。 -
我们将省略
insert_batch和insert_allocation的细节；重点是创建几个批次，并且对于我们感兴趣的批次，使其分配到一个现有的订单行。 -
这就是我们在此处验证的内容。第一个
assert ==检查类型是否匹配，以及引用是否相同（因为，正如您所记得的，Batch是一个实体，并且我们为其自定义了 __eq__）。 -
因此，我们还明确检查了其主要属性，包括
._allocations，这是一个 PythonOrderLine值对象的集合。
您是否费力地为每个模型编写测试是一个判断性决定。一旦您对一个类进行了创建/修改/保存测试，如果您所有的类都遵循类似的模式，您可能会很高兴继续进行其他类的最小往返测试，甚至什么都不做。在我们的例子中，设置 ._allocations 集合的 ORM 配置有点复杂，因此值得进行特定的测试。
您最终会得到类似这样的东西
class SqlAlchemyRepository(AbstractRepository):
def __init__(self, session):
self.session = session
def add(self, batch):
self.session.add(batch)
def get(self, reference):
return self.session.query(model.Batch).filter_by(reference=reference).one()
def list(self):
return self.session.query(model.Batch).all()现在我们的 Flask 端点可能看起来像这样
@flask.route.gubbins
def allocate_endpoint():
batches = SqlAlchemyRepository.list()
lines = [
OrderLine(l['orderid'], l['sku'], l['qty'])
for l in request.params...
]
allocate(lines, batches)
session.commit()
return 201现在构建用于测试的伪造仓库变得微不足道！
这是仓库模式的最大好处之一
class FakeRepository(AbstractRepository):
def __init__(self, batches):
self._batches = set(batches)
def add(self, batch):
self._batches.add(batch)
def get(self, reference):
return next(b for b in self._batches if b.reference == reference)
def list(self):
return list(self._batches)因为它只是 set 的一个简单包装器，所以所有方法都是单行代码。
在测试中使用伪造仓库非常容易，并且我们有一个易于使用和推理的简单抽象
fake_repo = FakeRepository([batch1, batch2, batch3])您将在下一章中看到此伪造仓库的实际应用。
|
提示
|
为您的抽象构建伪造是获得设计反馈的绝佳方法：如果难以伪造，则抽象可能过于复杂。 |
什么是端口，什么是适配器（在 Python 中）？
我们不想在这里过多地纠缠于术语，因为我们想要关注的主要内容是依赖倒置，而您使用的技术的具体细节并不重要。此外，我们意识到不同的人使用略有不同的定义。
端口和适配器来自 OO 世界，我们坚持的定义是，端口是我们应用程序与我们希望抽象掉的任何事物之间的接口，而适配器是该接口或抽象背后的实现。
现在 Python 没有接口本身，因此，尽管通常很容易识别适配器，但定义端口可能更困难。如果您使用的是抽象基类，则它是端口。如果不是，则端口只是您的适配器符合并且您的核心应用程序期望的鸭子类型——正在使用的函数和方法名称，以及它们的参数名称和类型。
具体来说，在本章中，AbstractRepository 是端口，SqlAlchemyRepository 和 FakeRepository 是适配器。
总结
请记住 Rich Hickey 的名言，在每一章中，我们都会总结我们介绍的每种架构模式的成本和收益。我们要明确表示，我们并不是说每个应用程序都需要以这种方式构建；只有当应用程序和领域的复杂性使其值得投入时间和精力来添加这些额外的间接层时，才需要这样做。
考虑到这一点，仓库模式和持久性无知：权衡 展示了仓库模式和我们的持久性无知模型的一些优缺点。
| 优点 | 缺点 |
|---|---|
|
|
领域模型权衡作为图表 显示了基本论点：是的，对于简单的情况，解耦的领域模型比简单的 ORM/ActiveRecord 模式更难工作。[8]
|
提示
|
如果您的应用程序只是围绕数据库的简单 CRUD（创建-读取-更新-删除）包装器，那么您不需要领域模型或仓库。 |
但是，领域越复杂，在将自己从基础设施关注点中解放出来方面的投资就越能在进行更改的便利性方面得到回报。
我们的示例代码不够复杂，无法给出图中右侧的更多提示，但是提示就在那里。例如，想象一下，如果我们有一天决定要将分配更改为驻留在 OrderLine 而不是 Batch 对象上：如果我们使用 Django，例如，我们将必须定义和考虑数据库迁移，然后才能运行任何测试。就像现在这样，因为我们的模型只是普通的旧 Python 对象，所以我们可以将 set() 更改为新属性，而无需考虑数据库，直到稍后。
您会想知道，我们如何实例化这些仓库，伪造的还是真实的？我们的 Flask 应用程序实际上会是什么样子？您将在下一个激动人心的部分，服务层模式 中找到答案。
但首先，简短的题外话。