buy the book ribbon

附录 D：使用 Django 的 Repository 和 Unit of Work 模式

假设你想使用 Django 而不是 SQLAlchemy 和 Flask。事情会变成什么样？首先要选择在哪里安装它。我们把它放在主分配代码旁边的单独包中

├── src
│   ├── allocation
│   │   ├── __init__.py
│   │   ├── adapters
│   │   │   ├── __init__.py
...
│   ├── djangoproject
│   │   ├── alloc
│   │   │   ├── __init__.py
│   │   │   ├── apps.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   └── views.py
│   │   ├── django_project
│   │   │   ├── __init__.py
│   │   │   ├── settings.py
│   │   │   ├── urls.py
│   │   │   └── wsgi.py
│   │   └── manage.py
│   └── setup.py
└── tests
    ├── conftest.py
    ├── e2e
    │   └── test_api.py
    ├── integration
    │   ├── test_repository.py
...
提示

此附录的代码在 appendix_django 分支 GitHub 上

git clone https://github.com/cosmicpython/code.git
cd code
git checkout appendix_django

代码示例紧接 [chapter_06_uow] 的结尾。

使用 Django 的 Repository 模式

我们使用了一个名为 pytest-django 的插件来帮助进行测试数据库管理。

重写第一个 repository 测试是一个最小的更改——只是用对 Django ORM/QuerySet 语言的调用重写了一些原始 SQL

第一个 repository 测试已调整 (tests/integration/test_repository.py)
from djangoproject.alloc import models as django_models


@pytest.mark.django_db
def test_repository_can_save_a_batch():
    batch = model.Batch("batch1", "RUSTY-SOAPDISH", 100, eta=date(2011, 12, 25))

    repo = repository.DjangoRepository()
    repo.add(batch)

    [saved_batch] = django_models.Batch.objects.all()
    assert saved_batch.reference == batch.reference
    assert saved_batch.sku == batch.sku
    assert saved_batch.qty == batch._purchased_quantity
    assert saved_batch.eta == batch.eta

第二个测试稍微复杂一些，因为它有分配，但它仍然由看起来很熟悉的 Django 代码组成

第二个 repository 测试更复杂 (tests/integration/test_repository.py)
@pytest.mark.django_db
def test_repository_can_retrieve_a_batch_with_allocations():
    sku = "PONY-STATUE"
    d_line = django_models.OrderLine.objects.create(orderid="order1", sku=sku, qty=12)
    d_batch1 = django_models.Batch.objects.create(
        reference="batch1", sku=sku, qty=100, eta=None
    )
    d_batch2 = django_models.Batch.objects.create(
        reference="batch2", sku=sku, qty=100, eta=None
    )
    django_models.Allocation.objects.create(line=d_line, batch=d_batch1)

    repo = repository.DjangoRepository()
    retrieved = repo.get("batch1")

    expected = model.Batch("batch1", sku, 100, eta=None)
    assert retrieved == expected  # Batch.__eq__ only compares reference
    assert retrieved.sku == expected.sku
    assert retrieved._purchased_quantity == expected._purchased_quantity
    assert retrieved._allocations == {
        model.OrderLine("order1", sku, 12),
    }

这是实际的 repository 最终看起来的样子

一个 Django repository (src/allocation/adapters/repository.py)
class DjangoRepository(AbstractRepository):
    def add(self, batch):
        super().add(batch)
        self.update(batch)

    def update(self, batch):
        django_models.Batch.update_from_domain(batch)

    def _get(self, reference):
        return (
            django_models.Batch.objects.filter(reference=reference)
            .first()
            .to_domain()
        )

    def list(self):
        return [b.to_domain() for b in django_models.Batch.objects.all()]

你可以看到，该实现依赖于 Django 模型具有一些自定义方法，用于与我们的领域模型相互转换。[1]

Django ORM 类上的自定义方法，用于与我们的领域模型相互转换

这些自定义方法看起来像这样

带有用于领域模型转换的自定义方法的 Django ORM (src/djangoproject/alloc/models.py)
from django.db import models
from allocation.domain import model as domain_model


class Batch(models.Model):
    reference = models.CharField(max_length=255)
    sku = models.CharField(max_length=255)
    qty = models.IntegerField()
    eta = models.DateField(blank=True, null=True)

    @staticmethod
    def update_from_domain(batch: domain_model.Batch):
        try:
            b = Batch.objects.get(reference=batch.reference)  #(1)
        except Batch.DoesNotExist:
            b = Batch(reference=batch.reference)  #(1)
        b.sku = batch.sku
        b.qty = batch._purchased_quantity
        b.eta = batch.eta  #(2)
        b.save()
        b.allocation_set.set(
            Allocation.from_domain(l, b)  #(3)
            for l in batch._allocations
        )

    def to_domain(self) -> domain_model.Batch:
        b = domain_model.Batch(
            ref=self.reference, sku=self.sku, qty=self.qty, eta=self.eta
        )
        b._allocations = set(
            a.line.to_domain()
            for a in self.allocation_set.all()
        )
        return b


class OrderLine(models.Model):
    #...

  1. 对于值对象，objects.get_or_create 可以工作，但对于实体，你可能需要显式的 try-get/except 来处理 upsert。[2]

  2. 我们在这里展示了最复杂的示例。如果你确实决定这样做，请注意会有样板代码！幸运的是，它不是很复杂的样板代码。

  3. 关系也需要一些仔细的自定义处理。

注意
 如 [chapter_02_repository] 中所述，我们使用依赖倒置。ORM (Django) 依赖于模型，而不是反过来。

使用 Django 的 Unit of Work 模式

测试没有太大变化

调整后的 UoW 测试 (tests/integration/test_uow.py)
def insert_batch(ref, sku, qty, eta):  #(1)
    django_models.Batch.objects.create(reference=ref, sku=sku, qty=qty, eta=eta)


def get_allocated_batch_ref(orderid, sku):  #(1)
    return django_models.Allocation.objects.get(
        line__orderid=orderid, line__sku=sku
    ).batch.reference


@pytest.mark.django_db(transaction=True)
def test_uow_can_retrieve_a_batch_and_allocate_to_it():
    insert_batch("batch1", "HIPSTER-WORKBENCH", 100, None)

    uow = unit_of_work.DjangoUnitOfWork()
    with uow:
        batch = uow.batches.get(reference="batch1")
        line = model.OrderLine("o1", "HIPSTER-WORKBENCH", 10)
        batch.allocate(line)
        uow.commit()

    batchref = get_allocated_batch_ref("o1", "HIPSTER-WORKBENCH")
    assert batchref == "batch1"


@pytest.mark.django_db(transaction=True)  #(2)
def test_rolls_back_uncommitted_work_by_default():
    ...

@pytest.mark.django_db(transaction=True)  #(2)
def test_rolls_back_on_error():
    ...

  1. 因为我们在这些测试中使用了小的辅助函数，所以测试的实际主体与使用 SQLAlchemy 时的主体几乎相同。

  2. 需要 pytest-django mark.django_db(transaction=True) 来测试我们的自定义事务/回滚行为。

并且实现非常简单，尽管我尝试了几次才找到哪个 Django 事务魔法的调用会起作用

为 Django 调整的 UoW (src/allocation/service_layer/unit_of_work.py)
class DjangoUnitOfWork(AbstractUnitOfWork):
    def __enter__(self):
        self.batches = repository.DjangoRepository()
        transaction.set_autocommit(False)  #(1)
        return super().__enter__()

    def __exit__(self, *args):
        super().__exit__(*args)
        transaction.set_autocommit(True)

    def commit(self):
        for batch in self.batches.seen:  #(3)
            self.batches.update(batch)  #(3)
        transaction.commit()  #(2)

    def rollback(self):
        transaction.rollback()  #(2)

  1. set_autocommit(False) 是告诉 Django 停止自动立即提交每个 ORM 操作并开始事务的最佳方法。

  2. 然后我们使用显式的回滚和提交。

  3. 一个难点：因为与 SQLAlchemy 不同，我们没有检测领域模型实例本身，所以 commit() 命令需要显式地遍历每个 repository 触摸过的所有对象，并将它们手动更新回 ORM。

API：Django 视图是适配器

Django 的 views.py 文件最终与旧的 flask_app.py 几乎相同，因为我们的架构意味着它是我们服务层（顺便说一句，它根本没有改变）周围的一个非常薄的包装器

Flask 应用 → Django 视图 (src/djangoproject/alloc/views.py)
os.environ["DJANGO_SETTINGS_MODULE"] = "djangoproject.django_project.settings"
django.setup()


@csrf_exempt
def add_batch(request):
    data = json.loads(request.body)
    eta = data["eta"]
    if eta is not None:
        eta = datetime.fromisoformat(eta).date()
    services.add_batch(
        data["ref"], data["sku"], data["qty"], eta,
        unit_of_work.DjangoUnitOfWork(),
    )
    return HttpResponse("OK", status=201)


@csrf_exempt
def allocate(request):
    data = json.loads(request.body)
    try:
        batchref = services.allocate(
            data["orderid"],
            data["sku"],
            data["qty"],
            unit_of_work.DjangoUnitOfWork(),
        )
    except (model.OutOfStock, services.InvalidSku) as e:
        return JsonResponse({"message": str(e)}, status=400)

    return JsonResponse({"batchref": batchref}, status=201)

为什么这一切如此困难？

好吧，它可以工作，但感觉比 Flask/SQLAlchemy 更费力。这是为什么呢？

低级别的主要原因是 Django 的 ORM 工作方式不同。我们没有 SQLAlchemy 经典映射器的等效物，因此我们的 ActiveRecord 和我们的领域模型不能是同一个对象。相反，我们必须在 repository 后面构建一个手动翻译层。这需要更多的工作（尽管一旦完成，持续的维护负担应该不会太高）。

因为 Django 与数据库紧密耦合，所以你必须使用像 pytest-django 这样的辅助工具，并从第一行代码开始就仔细考虑测试数据库，这在我们刚开始使用纯领域模型时不必这样做。

但在更高的层面上，Django 如此出色的整个原因是它围绕着轻松构建 CRUD 应用程序并最大限度减少样板代码的甜蜜点而设计的。但我们这本书的全部重点是当你的应用程序不再是一个简单的 CRUD 应用程序时该怎么办。

在这一点上，Django 开始弊大于利。像 Django 管理员这样的东西，在你刚开始时非常棒，但如果你的应用程序的全部目的是围绕状态更改的工作流程构建一套复杂的规则和模型，那么它们就会变得非常危险。Django 管理员绕过了所有这些。

如果你已经有 Django 该怎么办

那么，如果你想将本书中的一些模式应用于 Django 应用程序，你应该怎么做？我们会说以下几点

  • Repository 和 Unit of Work 模式将需要大量工作。它们在短期内将为你带来的主要好处是更快的单元测试，因此评估一下这种好处是否值得你付出努力。从长远来看，它们将你的应用程序与 Django 和数据库解耦，因此如果你预计想要从这两者中的任何一个迁移出来，Repository 和 UoW 是一个好主意。

  • 如果你在 views.py 中看到大量重复代码，那么服务层模式可能会引起你的兴趣。它可以很好地将你的用例与你的 Web 端点分开考虑。

  • 你仍然可以在理论上使用 Django 模型进行 DDD 和领域建模，尽管它们与数据库紧密耦合；你可能会因迁移而放慢速度，但这不应该是致命的。因此，只要你的应用程序不太复杂，并且你的测试速度不太慢，你或许可以从胖模型方法中获得一些好处：尽可能将更多的逻辑下推到你的模型中，并应用诸如实体、值对象和聚合之类的模式。但是，请参阅以下注意事项。

话虽如此，Django 社区的说法 是，人们发现胖模型方法本身也遇到了可扩展性问题，尤其是在管理应用程序之间的相互依赖关系方面。在这些情况下，提取出一个业务逻辑或领域层来位于你的视图和表单与你的 models.py 之间，然后你可以尽可能地保持最小化，这有很多道理。

沿途的步骤

假设你正在处理一个 Django 项目，你不确定它是否会变得足够复杂以至于需要我们推荐的模式，但你仍然想采取一些步骤来让你的生活更轻松，无论是在中期还是在你以后想迁移到我们的一些模式时。请考虑以下几点

  • 我们听到的一个建议是从第一天开始就在每个 Django 应用程序中放入一个 logic.py。这为你提供了一个放置业务逻辑的地方，并使你的表单、视图和模型免受业务逻辑的影响。它可以成为以后迁移到完全解耦的领域模型和/或服务层的垫脚石。

  • 业务逻辑层可能一开始与 Django 模型对象一起工作，并且稍后才会完全与框架解耦，并在纯 Python 数据结构上工作。

  • 对于读取端，你可以通过将读取放在一个地方，避免 ORM 调用遍布各处，从而获得 CQRS 的一些好处。

  • 在分离读取模块和领域逻辑模块时，可能值得将自己与 Django 应用程序层次结构解耦。业务关注点将贯穿它们。

注意
 我们要向 David Seddon 和 Ashia Zawaduk 致谢，感谢他们讨论了本附录中的一些想法。他们尽力阻止我们说出任何关于我们个人经验不足的主题的真正愚蠢的话，但他们可能失败了。

有关处理现有应用程序的更多想法和实际生活经验，请参阅 尾声。

<< 上一章 - 附录 C：替换基础设施：使用 CSV 完成所有操作 下一章 - 附录 E：验证 >>
1. DRY-Python 项目人员构建了一个名为 mappers 的工具，它看起来可能有助于最大限度地减少此类事情的样板代码。
2. @mr-bo-jangles 建议你或许可以使用 update_or_create，但这超出了我们的 Django-fu。