2 存储库模式 Repository Pattern

原文: https://www.cosmicpython.com/book/chapter_02_repository.html

以下大部分来源于机翻

是时候使用依赖性反转原则作为将我们的核心逻辑与基础设施问题脱钩的一种方式。

我们将引入存储库模式，一种简化的数据存储抽象，这让模型层与数据层解耦。我们将举一个具体的例子，说明这种简化的抽象如何通过隐藏数据库的复杂性来使我们的系统更具可测试性。

下图展示了使用repository pattern使用前后的区别;我们在领域和数据库之间创建了一个Rpository对象.

持久化领域模型

在领域模型中，我们构建了一个简单的域模型，可以将订单分配给批量库存。我们很容易根据此代码编写测试，因为没有任何依赖项或基础设施需要设置。如果我们需要运行数据库或API并创建测试数据，我们的测试将更难编写和维护。

可悲的是，在某个时候，我们需要将我们完美的小模型交到用户手中，并与电子表格、网络浏览器和比赛条件的现实世界作斗争。在接下来的几章中，我们将研究如何将理想化的域模型连接到外部状态。

我们期望以敏捷的方式工作，因此我们的首要任务是尽快获得MVP的可行产品。就我们而言，这将是一个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

我们需要一种方法从数据库中检索批处理信息，并从中实例化我们的域模型对象，我们还需要一种将它们保存回数据库的方法

将依赖反转原则应用于数据访问

正如导言中提到的常见3层架构, UI层、逻辑层、数据层.
Django的Model-View-Template结构也密切相关，Model-View-Controller（MVC）也是如此。无论如何，目的是保持分层隔离（这是一件好事），并且每个图层只依赖于它下面的图层。

但我们希望我们的域模型没有任何依赖性. 我们不希望基础设施关注领域模型，这会阻碍我们的单元测试或限制我们应用变更的能力.

相反，正如导言中讨论的那样，我们会认为我们的模型在“内部”，依赖关系向内流向, 这就是人们常说的洋葱架构.

https://www.cosmicpython.com/book/images/apwp_0202.png

回顾一下 我们的领域模型

让我们回顾一下我们的领域模型:分配是将OrderLine链接到Batch。我们将分配作为一个集合存储在Batch对象上。

https://www.cosmicpython.com/book/images/apwp_0103.png

让我们看看如何将其转换为关系数据库。

“正常”ORM方式：模型依赖于ORM

现在，您的团队成员不太可能亲自处理自己的SQL查询。相反，您几乎肯定会使用某种框架来根据模型对象为您生成SQL。

这些框架被称为对象-关系映射器(orm)，因为它们的存在是为了弥合对象和领域建模世界与数据库和关系代数世界之间的概念鸿沟。

ORM给我们的最重要的东西是对持久化的无感:即我们花哨的领域模型不需要知道数据是如何加载或持久的。这有助于保持我们的领域不直接依赖于特定的数据库技术

但是如果您遵循典型的SQLAlchemy教程，您将得到如下内容:

orm.py

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所称的classical mapping:

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)

1. ORM导入(或“依赖”或“了解”)域模型，而不是反过来。
2. 我们使用SQLAlchemy的抽象定义数据库表和列
3. 调用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)]

1. 如果没有使用过pytest，则需要解释此测试的会话参数。对于本书的目的，您不需要担心pytest或它的fixture的细节，但简短的解释是，您可以将测试的公共依赖项定义为“fixture”，pytest将通过查看它们的函数参数将它们注入到需要它们的测试中。在本例中，它是一个SQLAlchemy数据库会话。

您可能不会保留这些测试—您很快就会看到，一旦您采取了反转ORM和域模型的依赖关系，实现另一个称为Repository模式的抽象只需额外的一个小步骤，它将更容易编写测试，并将为稍后的测试提供一个简单的伪装接口。

但是我们已经实现了反转传统依赖关系的目标:域模型保持“纯粹”，不涉及基础设施问题。我们可以抛弃SQLAlchemy，使用不同的ORM，或者完全不同的持久性系统，域模型根本不需要更改。

这取决于您在域模型中所做的事情，特别是如果您偏离了OO范式，您可能会发现越来越难让ORM产生您需要的确切行为，您可能需要修改您的域模型正如架构决策经常发生的那样，您需要考虑取舍。正如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

仓库模式的介绍

Repository模式是持久存储的抽象。它假装我们所有的数据都在内存中，从而隐藏了数据访问的无聊细节。

如果我们的笔记本电脑有无限的内存，我们就不需要笨拙的数据库了。相反，我们可以随时使用我们的对象。那会是什么样子呢?

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()返回先前添加的项。我们严格坚持使用这些方法在我们的领域和服务层中进行数据访问。这种自我强加的简单性阻止我们将域模型与数据库耦合。

下面是我们仓库的抽象基类(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

1. Python提示:@abc。abstractmethod是使abc在Python中真正“工作”的唯一东西之一。Python将拒绝实例化一个没有实现父类中定义的所有抽象方法的类。
2. raise NotImplementedError很好，但它既不是必要的也不是充分的。事实上，如果你真的想，你的抽象方法可以有子类可以调用的真实行为。

抽象基类、鸭子类型和协议

我们在本书中使用抽象基类是出于教学目的:我们希望它们有助于解释存储库抽象的接口是什么。

在现实生活中，我们有时会发现自己从生产代码中删除abc，因为Python让忽略它们变得太容易了，它们最终没有得到维护，在最坏的情况下，会产生误导。在实践中，我们通常只依赖Python的duck类型来启用抽象。对于Pythonista来说，存储库是任何具有add(thing)和get(id)方法的对象。

另一种选择是PEP 544协议

每当我们在本书中介绍一个体系结构模式时，我们总是会问:“我们能从中得到什么?”我们要花多少钱?”

通常，至少，我们将引入一个额外的抽象层，尽管我们可能希望它将总体上减少复杂性，但它确实增加了局部的复杂性，并且它在移动部件的原始数量和持续维护方面有成本。

但是，如果您已经开始采用DDD和依赖倒置的方法，那么Repository模式可能是本书中最简单的选择之一。就我们的代码而言，我们实际上只是将SQLAlchemy抽象(session.query(Batch))转换为我们设计的另一个抽象(batches_repo.get)。

每次添加想要检索的新域对象时，我们必须在存储库类中编写几行代码，但作为回报，我们得到了对我们控制的存储层的简单抽象。Repository模式可以很容易地对存储内容的方式进行基本更改(参见[appendix_csvs])，并且我们将看到，在单元测试中很容易伪造出来。

此外，Repository模式在DDD世界中非常常见，如果您是从Java和c#世界来到Python的程序员，他们很可能会认出它。存储库模式说明了该模式。

https://www.cosmicpython.com/book/images/apwp_0205.png

和往常一样，我们从一个测试开始。这可能会被归类为集成测试，因为我们正在检查我们的代码(存储库)是否正确地与数据库集成;因此，测试倾向于在我们自己的代码上混合使用原始SQL和调用和断言

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

1. repo.add() 是这里要测试的内容
2. 我们将.commit()保持在存储库之外，并使其由调用者负责。这种做法有利有弊;当我们读到uow时，我们的一些理由会变得更清楚。
3. 我们用原始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),
    }

1. 这将测试读取端，因此原始SQL正在准备由repo.get()读取的数据。
2. 我们将不介绍insert_batch和insert_allocation的细节;重点是创建两个Batches，并且，对于我们感兴趣的Batches，将一个现有的订单分配给它。
3. 这就是我们要验证的。第一个assert ==检查类型是否匹配，以及引用是否相同(因为，正如您所记得的，Batch是一个实体，我们有一个自定义的__eq__用于它)。
4. 因此，我们还显式检查它的主要属性，包括._allocations，这是一个Python的Order集合

是否煞费苦心地为每个模型编写测试是一种判断。一旦对一个类进行了创建/修改/保存测试，您可能很乐意继续进行其他类的创建/修改/保存测试，如果它们都遵循类似的模式，那么只需进行最少的往返测试，甚至不进行任何测试。在我们的例子中，设置._allocation集的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

为测试构建一个假的存储库现在是微不足道的!

下面是Repository模式的最大好处之一:

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)

因为它是一个简单的集合包装器，所有的方法都是一行程序

使用伪仓库进行测试时如此容易,简单的抽象更容易使用和理解.

fake_repo = FakeRepository([batch1, batch2, batch3])

在Python中什么是 port、什么是Adapter?

我们不想在这里过多地讨论术语，因为我们主要想关注的是依赖性反转，使用的技术的具体细节并不重要。另外，我们知道不同的人使用的定义略有不同。

端口和适配器来自于OO世界，我们坚持的定义是，端口是应用程序和任何我们希望抽象的东西之间的接口，而适配器是接口或抽象背后的实现

现在Python本身没有接口，所以尽管通常很容易识别适配器，但定义端口可能比较困难。如果您正在使用抽象基类，那就是端口。如果不是，则端口只是适配器所遵循的鸭类型，也是核心应用程序所期望的—所使用的函数和方法名称，以及它们的参数名称和类型。

具体地说，在本章中，AbstractRepository是端口，SqlAlchemyRepository和FakeRepository是适配器。

总结

记住Rich Hickey的名言，在每一章中，我们总结了所介绍的每种体系结构模式的成本和收益。我们想要明确的是，我们并不是说每个应用程序都需要这样构建;只有在某些情况下，应用程序和领域的复杂性才值得投入时间和精力来添加这些额外的间接层。

记住这一点，仓库模式和持久化无感:要权衡库模式和我们的持久性无感的一些优点和缺点。

https://www.cosmicpython.com/book/images/apwp_0206.png

仓库模式概览

1. ORM中应用依赖反转
2. 仓库模式是对存储层的简单抽象