chapter_13_dependency_injection.asciidoc

Dependency Injection (and Bootstrapping)

依赖注入（以及引导启动）

Dependency injection (DI) is regarded with suspicion in the Python world. And we’ve managed just fine without it so far in the example code for this book!

依赖注入（DI）在 Python 世界中常常被视为一种值得怀疑的实践。然而，到目前为止，在本书的示例代码中，我们已经做得 相当不错，并且没有使用它！

In this chapter, we’ll explore some of the pain points in our code that lead us to consider using DI, and we’ll present some options for how to do it, leaving it to you to pick which you think is most Pythonic.

在本章中，我们将探讨代码中的一些痛点，这些痛点促使我们考虑使用依赖注入（DI）。同时，我们将介绍几种实现方法， 并留给你来选择你认为最符合 Python 风格的方法。

We’ll also add a new component to our architecture called bootstrap.py; it will be in charge of dependency injection, as well as some other initialization stuff that we often need. We’ll explain why this sort of thing is called a composition root in OO languages, and why bootstrap script is just fine for our purposes.

我们还将在我们的架构中添加一个新的组件，名为 bootstrap.py；它将负责依赖注入，以及一些我们经常需要的其他初始化工作。 我们会解释为什么在面向对象（OO）语言中，这类东西被称为 组合根，以及为什么对于我们的目的来说，称为 引导脚本 就完全可以。

Without bootstrap: entrypoints do a lot（没有引导程序：入口点负责大量工作） shows what our app looks like without a bootstrapper: the entrypoints do a lot of initialization and passing around of our main dependency, the UoW.

Without bootstrap: entrypoints do a lot（没有引导程序：入口点负责大量工作） 展示了我们在没有引导程序（bootstrapper）时的应用程序结构：入口点（entrypoints）负责大量的 初始化工作，并在各处传递我们的主要依赖——工作单元模式。

Tip	If you haven’t already, it’s worth reading [chapter_03_abstractions] before continuing with this chapter, particularly the discussion of functional versus object-oriented dependency management. 如果你还没有阅读过 [chapter_03_abstractions]，那么在继续阅读本章之前，值得先读一读，尤其是其中关于函数式与面向对象的依赖管理的讨论。

Figure 1. Without bootstrap: entrypoints do a lot（没有引导程序：入口点负责大量工作）

Tip

The code for this chapter is in the chapter_13_dependency_injection branch on GitHub: 本章的代码位于 chapter_13_dependency_injection 分支 在 GitHub 上： git clone https://github.com/cosmicpython/code.git cd code git checkout chapter_13_dependency_injection # or to code along, checkout the previous chapter: git checkout chapter_12_cqrs

Bootstrap takes care of all that in one place（引导程序统一处理所有这些） shows our bootstrapper taking over those responsibilities.

Bootstrap takes care of all that in one place（引导程序统一处理所有这些） 展示了我们的引导程序（bootstrapper）接管了这些职责。

Figure 2. Bootstrap takes care of all that in one place（引导程序统一处理所有这些）

Implicit Versus Explicit Dependencies

隐式依赖与显式依赖

Depending on your particular brain type, you may have a slight feeling of unease at the back of your mind at this point. Let’s bring it out into the open. We’ve shown you two ways of managing dependencies and testing them.

根据你的思维类型，你此刻可能会在内心深处隐约感到一丝不安。让我们把这个问题摆到台面上来讨论。我们已经向你展示了两种管理依赖和测试依赖的方法。

For our database dependency, we’ve built a careful framework of explicit dependencies and easy options for overriding them in tests. Our main handler functions declare an explicit dependency on the UoW:

对于我们的数据库依赖，我们构建了一个精心设计的框架，提供了显式依赖以及在测试中轻松覆盖它们的选项。我们的主要处理函数明确声明了对工作单元的依赖：

Example 1. Our handlers have an explicit dependency on the UoW (src/allocation/service_layer/handlers.py)（我们的处理程序显式依赖于 UoW）

def allocate(
    cmd: commands.Allocate,
    uow: unit_of_work.AbstractUnitOfWork,
):

And that makes it easy to swap in a fake UoW in our service-layer tests:

这使得在我们的服务层测试中轻松替换为一个假的工作单元成为可能：

Example 2. Service-layer tests against a fake UoW: (tests/unit/test_services.py)（用伪造的工作单元进行服务层测试）

    uow = FakeUnitOfWork()
    messagebus.handle([...], uow)

The UoW itself declares an explicit dependency on the session factory:

工作单元本身明确声明了对会话工厂（session factory）的依赖：

Example 3. The UoW depends on a session factory (src/allocation/service_layer/unit_of_work.py)（工作单元依赖于会话工厂）

class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
    def __init__(self, session_factory=DEFAULT_SESSION_FACTORY):
        self.session_factory = session_factory
        ...

We take advantage of it in our integration tests to be able to sometimes use SQLite instead of Postgres:

我们在集成测试中利用了这一点，以便有时可以使用 SQLite 替代 Postgres：

Example 4. Integration tests against a different DB (tests/integration/test_uow.py)（针对不同数据库的集成测试）

def test_rolls_back_uncommitted_work_by_default(sqlite_session_factory):
    uow = unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory)  #(1)

1. Integration tests swap out the default Postgres session_factory for a SQLite one. 集成测试将默认的 Postgres session_factory 替换为 SQLite 的会话工厂。

Aren’t Explicit Dependencies Totally Weird and Java-y?

显式依赖难道不是非常奇怪且像 Java 的东西吗？

If you’re used to the way things normally happen in Python, you’ll be thinking all this is a bit weird. The standard way to do things is to declare our dependency implicitly by simply importing it, and then if we ever need to change it for tests, we can monkeypatch, as is Right and True in dynamic languages:

如果你习惯了 Python 中的常规做法，你可能会觉得这一切都有点奇怪。标准的做法是通过简单地导入来隐式声明依赖，然后如果我们需要在测试中改变它， 可以使用猴子补丁（monkeypatch），这在动态语言中是正确且合乎常理的：

Example 5. Email sending as a normal import-based dependency (src/allocation/service_layer/handlers.py)（电子邮件发送作为基于普通导入的依赖项）

from allocation.adapters import email, redis_eventpublisher  #(1)
...

def send_out_of_stock_notification(
    event: events.OutOfStock,
    uow: unit_of_work.AbstractUnitOfWork,
):
    email.send(  #(2)
        "stock@made.com",
        f"Out of stock for {event.sku}",
    )

1. Hardcoded import 硬编码导入

2. Calls specific email sender directly 直接调用特定的电子邮件发送器

Why pollute our application code with unnecessary arguments just for the sake of our tests? mock.patch makes monkeypatching nice and easy:

为什么仅仅为了测试而用不必要的参数污染我们的应用程序代码呢？mock.patch 让猴子补丁变得简单方便：

Example 6. mock dot patch, thank you Michael Foord (tests/unit/test_handlers.py)（mock.patch，感谢 Michael Foord）

    with mock.patch("allocation.adapters.email.send") as mock_send_mail:
        ...

The trouble is that we’ve made it look easy because our toy example doesn’t send real email (email.send_mail just does a print), but in real life, you’d end up having to call mock.patch for every single test that might cause an out-of-stock notification. If you’ve worked on codebases with lots of mocks used to prevent unwanted side effects, you’ll know how annoying that mocky boilerplate gets.

问题在于，我们让这一切看起来很简单，是因为我们的示例程序并未真正发送邮件（email.send_mail 只是执行一个 print 操作），但在现实情况下， 你最终不得不为 每一个可能触发缺货通知的测试 调用 mock.patch。如果你曾在代码库中处理过许多用于防止不必要副作用的 mock， 你会知道这些 mock 带来的模板化代码有多么令人厌烦。

And you’ll know that mocks tightly couple us to the implementation. By choosing to monkeypatch email.send_mail, we are tied to doing import email, and if we ever want to do from email import send_mail, a trivial refactor, we’d have to change all our mocks.

你还会知道，mock 会将我们与实现紧密耦合。通过选择对 email.send_mail 进行猴子补丁（monkeypatch）， 我们就绑定到了 import email 的用法上。如果我们哪天想改成 from email import send_mail 这样一个看似简单的重构， 就必须修改所有的 mock。

So it’s a trade-off. Yes, declaring explicit dependencies is unnecessary, strictly speaking, and using them would make our application code marginally more complex. But in return, we’d get tests that are easier to write and manage.

所以这是一个权衡问题。严格来说，声明显式依赖并不是必须的，使用它们确实会让我们的应用程序代码略微复杂一些。 但作为回报，我们会得到更容易编写和管理的测试代码。

On top of that, declaring an explicit dependency is an example of the dependency inversion principle—rather than having an (implicit) dependency on a specific detail, we have an (explicit) dependency on an abstraction:

除此之外，声明显式依赖是依赖倒置原则的一个实例——与其对某个 具体 细节有（隐式的）依赖，不如对一个 抽象 有（显式的）依赖：

Explicit is better than implicit.

显式优于隐式。

— The Zen of Python

Example 7. The explicit dependency is more abstract (src/allocation/service_layer/handlers.py)（显式依赖更加抽象）

def send_out_of_stock_notification(
    event: events.OutOfStock,
    send_mail: Callable,
):
    send_mail(
        "stock@made.com",
        f"Out of stock for {event.sku}",
    )

But if we do change to declaring all these dependencies explicitly, who will inject them, and how? So far, we’ve really been dealing with only passing the UoW around: our tests use FakeUnitOfWork, while Flask and Redis eventconsumer entrypoints use the real UoW, and the message bus passes them onto our command handlers. If we add real and fake email classes, who will create them and pass them on?

但是，如果我们确实改为显式声明所有这些依赖，那么谁来注入它们，又该如何注入呢？到目前为止，我们实际上只是处理了工作单元的传递： 我们的测试中使用 FakeUnitOfWork，而 Flask 和 Redis 的事件消费者入口点使用真正的工作单元，消息总线将它们传递给命令处理器。 如果我们添加真实和假的电子邮件类，那么谁来创建它们并传递下去呢？

It needs to happen as early as possible in the process lifecycle, so the most obvious place is in our entrypoints. That would mean extra (duplicated) cruft in Flask and Redis, and in our tests. And we’d also have to add the responsibility for passing dependencies around to the message bus, which already has a job to do; it feels like a violation of the SRP.

这种注入需要尽早发生在进程生命周期中，因此最明显的位置是在我们的入口点。这意味着在 Flask 和 Redis 以及测试中都会出现额外的（重复的）累赘。 同时，我们还需要将传递依赖的责任添加到消息总线上，而消息总线本身已经有自己的职责；这么做感觉违反了单一职责原则（SRP）。

Instead, we’ll reach for a pattern called Composition Root (a bootstrap script to you and me),^[1] and we’ll do a bit of "manual DI" (dependency injection without a framework). See Bootstrapper between entrypoints and message bus（引导程序位于入口点与消息总线之间）.^[2]

相反，我们将使用一种被称为 组合根（在你我看来就是一个引导脚本）的模式，脚注：[因为 Python 不是一种“纯”面向对象语言， Python 开发者并不一定习惯需要“组合”一组对象来构建一个可运行的应用程序。我们通常只是选择一个入口点，然后从上到下运行代码。] 并且我们将进行一些“手动依赖注入”（不用框架实现的依赖注入）。请参见 Bootstrapper between entrypoints and message bus（引导程序位于入口点与消息总线之间）。 脚注：[Mark Seemann 将这种做法称为 纯依赖注入（Pure DI） 或称之为 原生依赖注入（Vanilla DI）。]

Figure 3. Bootstrapper between entrypoints and message bus（引导程序位于入口点与消息总线之间）

[ditaa, apwp_1303]

+---------------+
|  Entrypoints  |
| (Flask/Redis) |
+---------------+
        |
        | call
        V
 /--------------\
 |              |  prepares handlers with correct dependencies injected in
 | Bootstrapper |  (test bootstrapper will use fakes, prod one will use real)
 |              |
 \--------------/
        |
        | pass injected handlers to
        V
/---------------\
|  Message Bus  |
+---------------+
        |
        | dispatches events and commands to injected handlers
        |
        V

Preparing Handlers: Manual DI with Closures and Partials

准备处理器：使用闭包和偏函数的手动依赖注入（Manual DI）

One way to turn a function with dependencies into one that’s ready to be called later with those dependencies already injected is to use closures or partial functions to compose the function with its dependencies:

将一个带有依赖的函数转换成一个依赖 已注入 并准备好被稍后调用的函数的一种方法是使用闭包或偏函数，将函数与其依赖组合起来：

Example 8. Examples of DI using closures or partial functions（使用闭包或偏函数实现依赖注入的示例）

# existing allocate function, with abstract uow dependency
def allocate(
    cmd: commands.Allocate,
    uow: unit_of_work.AbstractUnitOfWork,
):
    line = OrderLine(cmd.orderid, cmd.sku, cmd.qty)
    with uow:
        ...

# bootstrap script prepares actual UoW

def bootstrap(..):
    uow = unit_of_work.SqlAlchemyUnitOfWork()

    # prepare a version of the allocate fn with UoW dependency captured in a closure
    allocate_composed = lambda cmd: allocate(cmd, uow)

    # or, equivalently (this gets you a nicer stack trace)
    def allocate_composed(cmd):
        return allocate(cmd, uow)

    # alternatively with a partial
    import functools
    allocate_composed = functools.partial(allocate, uow=uow)  #(1)

# later at runtime, we can call the partial function, and it will have
# the UoW already bound
allocate_composed(cmd)

1. The difference between closures (lambdas or named functions) and functools.partial is that the former use late binding of variables, which can be a source of confusion if any of the dependencies are mutable. 闭包（lambda 或命名函数）与 functools.partial 的区别在于，前者使用 延迟绑定变量， 如果某些依赖是可变的，这可能成为混淆的来源。

Here’s the same pattern again for the send_out_of_stock_notification() handler, which has different dependencies:

以下是针对 send_out_of_stock_notification() 处理器的相同模式示例，不过它具有不同的依赖：

Example 9. Another closure and partial functions example（另一个关于闭包和偏函数的示例）

def send_out_of_stock_notification(
    event: events.OutOfStock,
    send_mail: Callable,
):
    send_mail(
        "stock@made.com",
        ...


# prepare a version of the send_out_of_stock_notification with dependencies
sosn_composed  = lambda event: send_out_of_stock_notification(event, email.send_mail)

...
# later, at runtime:
sosn_composed(event)  # will have email.send_mail already injected in

An Alternative Using Classes

使用类的另一种方法

Closures and partial functions will feel familiar to people who’ve done a bit of functional programming. Here’s an alternative using classes, which may appeal to others. It requires rewriting all our handler functions as classes, though:

闭包和偏函数对于做过一些函数式编程的人来说会比较熟悉。这里提供了一种使用类的替代方法，这可能会吸引其他人。 不过，这需要将我们所有的处理器函数重写为类：

Example 10. DI using classes（使用类进行依赖注入）

# we replace the old `def allocate(cmd, uow)` with:

class AllocateHandler:
    def __init__(self, uow: unit_of_work.AbstractUnitOfWork):  #(2)
        self.uow = uow

    def __call__(self, cmd: commands.Allocate):  #(1)
        line = OrderLine(cmd.orderid, cmd.sku, cmd.qty)
        with self.uow:
            # rest of handler method as before
            ...

# bootstrap script prepares actual UoW
uow = unit_of_work.SqlAlchemyUnitOfWork()

# then prepares a version of the allocate fn with dependencies already injected
allocate = AllocateHandler(uow)

...
# later at runtime, we can call the handler instance, and it will have
# the UoW already injected
allocate(cmd)

1. The class is designed to produce a callable function, so it has a __call__ method. 该类被设计为生成一个可调用的函数，因此它有一个 __call__ 方法。

2. But we use the init to declare the dependencies it requires. This sort of thing will feel familiar if you’ve ever made class-based descriptors, or a class-based context manager that takes arguments. 但是我们使用 init 方法声明它所需要的依赖。如果你曾经实现过基于类的描述符或带参数的基于类的上下文管理器， 这种方式会让你感到熟悉。

Use whichever you and your team feel more comfortable with.

使用你和你的团队感到更舒适的方式即可。

A Bootstrap Script

引导脚本

We want our bootstrap script to do the following:

我们希望引导脚本完成以下任务：

1. Declare default dependencies but allow us to override them 声明默认依赖，但允许我们覆盖它们

2. Do the "init" stuff that we need to get our app started 完成启动我们的应用程序所需的“初始化”工作

3. Inject all the dependencies into our handlers 将所有依赖注入到我们的处理器中

4. Give us back the core object for our app, the message bus 将应用程序的核心对象——消息总线，返回给我们

Here’s a first cut:

以下是初步版本：

Example 11. A bootstrap function (src/allocation/bootstrap.py)（引导函数）

def bootstrap(
    start_orm: bool = True,  #(1)
    uow: unit_of_work.AbstractUnitOfWork = unit_of_work.SqlAlchemyUnitOfWork(),  #(2)
    send_mail: Callable = email.send,
    publish: Callable = redis_eventpublisher.publish,
) -> messagebus.MessageBus:

    if start_orm:
        orm.start_mappers()  #(1)

    dependencies = {"uow": uow, "send_mail": send_mail, "publish": publish}
    injected_event_handlers = {  #(3)
        event_type: [
            inject_dependencies(handler, dependencies)
            for handler in event_handlers
        ]
        for event_type, event_handlers in handlers.EVENT_HANDLERS.items()
    }
    injected_command_handlers = {  #(3)
        command_type: inject_dependencies(handler, dependencies)
        for command_type, handler in handlers.COMMAND_HANDLERS.items()
    }

    return messagebus.MessageBus(  #(4)
        uow=uow,
        event_handlers=injected_event_handlers,
        command_handlers=injected_command_handlers,
    )

1. orm.start_mappers() is our example of initialization work that needs to be done once at the beginning of an app. Another common example is setting up the logging module. orm.start_mappers() 是一个需要在应用程序启动时执行一次的初始化工作的示例。另一个常见的示例是设置 logging 模块。

2. We can use the argument defaults to define what the normal/production defaults are. It’s nice to have them in a single place, but sometimes dependencies have some side effects at construction time, in which case you might prefer to default them to None instead. 我们可以使用参数的默认值来定义正常/生产环境的默认配置。将它们集中在一个地方管理是很好的，但有时依赖在构造时可能会产生副作用， 在这种情况下，你或许更倾向于将默认值设置为 None。

3. We build up our injected versions of the handler mappings by using a function called inject_dependencies(), which we’ll show next. 我们通过一个名为 inject_dependencies() 的函数构建注入依赖后的处理器映射版本，我们将在接下来展示这个函数。

4. We return a configured message bus ready for use. 我们返回一个配置好的消息总线，准备好供使用。

Here’s how we inject dependencies into a handler function by inspecting it:

以下是通过检查处理器函数来向其注入依赖的方法：

Example 12. DI by inspecting function signatures (src/allocation/bootstrap.py)（通过检查函数签名进行依赖注入）

def inject_dependencies(handler, dependencies):
    params = inspect.signature(handler).parameters  #(1)
    deps = {
        name: dependency
        for name, dependency in dependencies.items()  #(2)
        if name in params
    }
    return lambda message: handler(message, **deps)  #(3)

1. We inspect our command/event handler’s arguments. 我们检查命令/事件处理器的参数。

2. We match them by name to our dependencies. 我们通过名称将它们与我们的依赖进行匹配。

3. We inject them as kwargs to produce a partial. 我们将它们作为关键字参数（kwargs）注入，以生成一个偏函数（partial）。

Even-More-Manual DI with Less Magic（更手动化、更少魔法的依赖注入）

If you’re finding the preceding inspect code a little harder to grok, this even simpler version may appeal to you.

如果你觉得前面的 inspect 代码有点难以理解，那么这个更简单的版本可能更适合你。

Harry wrote the code for inject_dependencies() as a first cut of how to do "manual" dependency injection, and when he saw it, Bob accused him of overengineering and writing his own DI framework.

Harry 编写了 inject_dependencies() 的代码，作为实现“手动”依赖注入的初步尝试，而当 Bob 看到它时，指责他过度设计，并且在写他自己的 DI 框架。

It honestly didn’t even occur to Harry that you could do it any more plainly, but you can, like this:

Harry 老实说完全没想到还可以用更简单的方式来实现，但事实上是可以的，像这样：

Example 13. Manually creating partial functions inline (src/allocation/bootstrap.py)（手动内联创建部分函数）

    injected_event_handlers = {
        events.Allocated: [
            lambda e: handlers.publish_allocated_event(e, publish),
            lambda e: handlers.add_allocation_to_read_model(e, uow),
        ],
        events.Deallocated: [
            lambda e: handlers.remove_allocation_from_read_model(e, uow),
            lambda e: handlers.reallocate(e, uow),
        ],
        events.OutOfStock: [
            lambda e: handlers.send_out_of_stock_notification(e, send_mail)
        ],
    }
    injected_command_handlers = {
        commands.Allocate: lambda c: handlers.allocate(c, uow),
        commands.CreateBatch: lambda c: handlers.add_batch(c, uow),
        commands.ChangeBatchQuantity: \
            lambda c: handlers.change_batch_quantity(c, uow),
    }

Harry says he couldn’t even imagine writing out that many lines of code and having to look up that many function arguments manually. It would be a perfectly viable solution, though, since it’s only one line of code or so per handler you add. Even if you have dozens of handlers, it wouldn’t be much of maintenance burden.

Harry 说他甚至无法想象要手写这么多行代码并手动查找这么多函数参数。然而，这确实是一个完全可行的解决方案，因为每增加一个处理器， 大约只需要一行代码。即使你有几十个处理器，这也不会带来太大的维护负担。

Our app is structured in such a way that we always want to do dependency injection in only one place, the handler functions, so this super-manual solution and Harry’s inspect()-based one will both work fine.

我们的应用程序结构设计使得我们始终只需要在一个地方——处理器函数中进行依赖注入， 因此这种完全手动解决方案和 Harry 基于 inspect() 的方案都可以很好地工作。

If you find yourself wanting to do DI in more things and at different times, or if you ever get into dependency chains (in which your dependencies have their own dependencies, and so on), you may get some mileage out of a "real" DI framework.

如果你发现自己想在更多的地方以及不同的时间执行依赖注入，或者你遇到了 依赖链（即你的依赖本身也有它们的依赖，以此类推）， 那么使用一个“真正的”依赖注入框架可能会有所帮助。

At MADE, we’ve used Inject in a few places, and it’s fine (although it makes Pylint unhappy). You might also check out Punq, as written by Bob himself, or the DRY-Python crew’s Dependencies.

在 MADE，我们在一些地方使用过 Inject，它表现得 还不错（尽管它会让 Pylint 不高兴）。 你也可以看看 Bob 自己写的 Punq， 或者 DRY-Python 团队的 Dependencies。

Message Bus Is Given Handlers at Runtime

消息总线在运行时分配处理器

Our message bus will no longer be static; it needs to have the already-injected handlers given to it. So we turn it from being a module into a configurable class:

我们的消息总线将不再是静态的；它需要接收已注入依赖的处理器。因此，我们将其从一个模块改为一个可配置的类：

Example 14. MessageBus as a class (src/allocation/service_layer/messagebus.py)（将 MessageBus 实现为一个类）

class MessageBus:  #(1)
    def __init__(
        self,
        uow: unit_of_work.AbstractUnitOfWork,
        event_handlers: Dict[Type[events.Event], List[Callable]],  #(2)
        command_handlers: Dict[Type[commands.Command], Callable],  #(2)
    ):
        self.uow = uow
        self.event_handlers = event_handlers
        self.command_handlers = command_handlers

    def handle(self, message: Message):  #(3)
        self.queue = [message]  #(4)
        while self.queue:
            message = self.queue.pop(0)
            if isinstance(message, events.Event):
                self.handle_event(message)
            elif isinstance(message, commands.Command):
                self.handle_command(message)
            else:
                raise Exception(f"{message} was not an Event or Command")

1. The message bus becomes a class…​ 消息总线变成了一个类…​

2. …​which is given its already-dependency-injected handlers. …​并接收已经完成依赖注入的处理器。

3. The main handle() function is substantially the same, with just a few attributes and methods moved onto self. 主要的 handle() 函数基本保持不变，只是将一些属性和方法移到了 self 上。

4. Using self.queue like this is not thread-safe, which might be a problem if you’re using threads, because the bus instance is global in the Flask app context as we’ve written it. Just something to watch out for. 像这样使用 self.queue 是非线程安全的，这可能会在使用线程时成为一个问题，因为在我们编写的代码中， 消息总线实例在 Flask 应用程序上下文中是全局的。这是需要注意的一点。

What else changes in the bus?

在消息总线中还有哪些变化？

Example 15. Event and command handler logic stays the same (src/allocation/service_layer/messagebus.py)（事件和命令处理逻辑保持不变）

    def handle_event(self, event: events.Event):
        for handler in self.event_handlers[type(event)]:  #(1)
            try:
                logger.debug("handling event %s with handler %s", event, handler)
                handler(event)  #(2)
                self.queue.extend(self.uow.collect_new_events())
            except Exception:
                logger.exception("Exception handling event %s", event)
                continue

    def handle_command(self, command: commands.Command):
        logger.debug("handling command %s", command)
        try:
            handler = self.command_handlers[type(command)]  #(1)
            handler(command)  #(2)
            self.queue.extend(self.uow.collect_new_events())
        except Exception:
            logger.exception("Exception handling command %s", command)
            raise

1. handle_event and handle_command are substantially the same, but instead of indexing into a static EVENT_HANDLERS or COMMAND_HANDLERS dict, they use the versions on self. handle_event 和 handle_command 基本保持不变，但它们不再索引到静态的 EVENT_HANDLERS 或 COMMAND_HANDLERS 字典， 而是使用 self 上的版本。

2. Instead of passing a UoW into the handler, we expect the handlers to already have all their dependencies, so all they need is a single argument, the specific event or command. 我们不再向处理器传递工作单元，而是期望处理器已经拥有它们所有的依赖，因此它们只需要一个参数，即特定的事件或命令。

Using Bootstrap in Our Entrypoints

在我们的入口点中使用引导程序（Bootstrap）

In our application’s entrypoints, we now just call bootstrap.bootstrap() and get a message bus that’s ready to go, rather than configuring a UoW and the rest of it:

在我们的应用程序入口点中，我们现在只需调用 bootstrap.bootstrap()，就能获得一个已配置好的消息总线，而无需手动配置工作单元和其他相关内容：

Example 16. Flask calls bootstrap (src/allocation/entrypoints/flask_app.py)（Flask 调用引导函数）

-from allocation import views
+from allocation import bootstrap, views

 app = Flask(__name__)
-orm.start_mappers()  #(1)
+bus = bootstrap.bootstrap()


 @app.route("/add_batch", methods=["POST"])
@@ -19,8 +16,7 @@ def add_batch():
     cmd = commands.CreateBatch(
         request.json["ref"], request.json["sku"], request.json["qty"], eta
     )
-    uow = unit_of_work.SqlAlchemyUnitOfWork()  #(2)
-    messagebus.handle(cmd, uow)
+    bus.handle(cmd)  #(3)
     return "OK", 201

1. We no longer need to call start_orm(); the bootstrap script’s initialization stages will do that. 我们不再需要调用 start_orm()；引导脚本的初始化阶段会处理这一点。

2. We no longer need to explicitly build a particular type of UoW; the bootstrap script defaults take care of it. 我们不再需要显式地构建特定类型的 UoW；引导脚本的默认设置会处理这一点。

3. And our message bus is now a specific instance rather than the global module.^[3] 我们的消息总线现在是一个特定的实例，而不是全局模块。脚注：[不过，它仍然是 flask_app 模块作用域内的一个全局变量，如果这样说得通的话。 如果你希望通过使用 Flask 测试客户端而不是像我们这样使用 Docker 来在进程内测试你的 Flask 应用，这可能会引发一些问题。如果遇到这种情况， 值得研究一下 Flask 应用工厂。]

Initializing DI in Our Tests

在我们的测试中初始化依赖注入

In tests, we can use bootstrap.bootstrap() with overridden defaults to get a custom message bus. Here’s an example in an integration test:

在测试中，我们可以使用 bootstrap.bootstrap() 并覆盖默认值以获取一个自定义消息总线。以下是一个集成测试中的示例：

Example 17. Overriding bootstrap defaults (tests/integration/test_views.py)（重写引导函数的默认设置）

@pytest.fixture
def sqlite_bus(sqlite_session_factory):
    bus = bootstrap.bootstrap(
        start_orm=True,  #(1)
        uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory),  #(2)
        send_mail=lambda *args: None,  #(3)
        publish=lambda *args: None,  #(3)
    )
    yield bus
    clear_mappers()


def test_allocations_view(sqlite_bus):
    sqlite_bus.handle(commands.CreateBatch("sku1batch", "sku1", 50, None))
    sqlite_bus.handle(commands.CreateBatch("sku2batch", "sku2", 50, today))
    ...
    assert views.allocations("order1", sqlite_bus.uow) == [
        {"sku": "sku1", "batchref": "sku1batch"},
        {"sku": "sku2", "batchref": "sku2batch"},
    ]

1. We do still want to start the ORM…​ 我们仍然需要启动 ORM…​

2. …​because we’re going to use a real UoW, albeit with an in-memory database. …​因为我们将使用一个真实的工作单元，尽管是基于内存的数据库。

3. But we don’t need to send email or publish, so we make those noops. 但我们不需要发送邮件或发布消息，所以我们将它们设为空操作（noops）。

In our unit tests, in contrast, we can reuse our FakeUnitOfWork:

相比之下，在我们的单元测试中，我们可以重用我们的 FakeUnitOfWork：

Example 18. Bootstrap in unit test (tests/unit/test_handlers.py)（单元测试中的引导函数）

def bootstrap_test_app():
    return bootstrap.bootstrap(
        start_orm=False,  #(1)
        uow=FakeUnitOfWork(),  #(2)
        send_mail=lambda *args: None,  #(3)
        publish=lambda *args: None,  #(3)
    )

1. No need to start the ORM…​ 不需要启动 ORM…​

2. …​because the fake UoW doesn’t use one. …​因为假的工作单元并不使用 ORM。

3. We want to fake out our email and Redis adapters too. 我们同样希望模拟（fake out）我们的电子邮件和 Redis 适配器。

So that gets rid of a little duplication, and we’ve moved a bunch of setup and sensible defaults into a single place.

这样可以减少一些重复工作，并且我们将大量的设置和合理的默认值集中到了一个地方。

Exercise for the Reader 1（读者练习 1）

Change all the handlers to being classes as per the DI using classes example, and amend the bootstrapper’s DI code as appropriate. This will let you know whether you prefer the functional approach or the class-based approach when it comes to your own projects.

将所有处理器更改为类，参考 使用类实现依赖注入 的示例，并相应修改引导程序的依赖注入（DI）代码。 通过这样做，你可以了解在你的项目中是倾向于函数式方法还是基于类的方法。

Building an Adapter "Properly": A Worked Example

“正确地”构建一个适配器：一个完整示例

To really get a feel for how it all works, let’s work through an example of how you might "properly" build an adapter and do dependency injection for it.

为了真正了解这一切是如何工作的，让我们通过一个示例来看看如何“正确地”构建一个适配器并为其执行依赖注入。

At the moment, we have two types of dependencies:

目前，我们有两种类型的依赖：

Example 19. Two types of dependencies (src/allocation/service_layer/messagebus.py)（两种类型的依赖）

    uow: unit_of_work.AbstractUnitOfWork,  #(1)
    send_mail: Callable,  #(2)
    publish: Callable,  #(2)

1. The UoW has an abstract base class. This is the heavyweight option for declaring and managing your external dependency. We’d use this for the case when the dependency is relatively complex. 工作单元有一个抽象基类。这是声明和管理外部依赖的重量级选项。我们会在依赖相对复杂的情况下使用这种方式。

2. Our email sender and pub/sub publisher are defined as functions. This works just fine for simple dependencies. 我们的电子邮件发送器和发布/订阅发布器被定义为函数。对于简单的依赖来说，这种方式完全够用。

Here are some of the things we find ourselves injecting at work:

以下是我们在工作中需要注入的一些内容：

• An S3 filesystem client 一个 S3 文件系统客户端

• A key/value store client 一个键/值存储客户端

• A requests session object 一个 requests 会话对象

Most of these will have more-complex APIs that you can’t capture as a single function: read and write, GET and POST, and so on.

其中大多数会有更加复杂的 API，无法用单个函数来概括：如读取和写入，GET 和 POST 等。

Even though it’s simple, let’s use send_mail as an example to talk through how you might define a more complex dependency.

尽管它很简单，但我们使用 send_mail 作为示例，来讨论如何定义一个更复杂的依赖。

Define the Abstract and Concrete Implementations

定义抽象实现和具体实现

We’ll imagine a more generic notifications API. Could be email, could be SMS, could be Slack posts one day.

我们可以设想一个更通用的通知 API。它可以是电子邮件，可能是短信，或者有一天是 Slack 消息。

Example 20. An ABC and a concrete implementation (src/allocation/adapters/notifications.py)（一个抽象基类 (ABC) 和一个具体实现）

class AbstractNotifications(abc.ABC):
    @abc.abstractmethod
    def send(self, destination, message):
        raise NotImplementedError

...

class EmailNotifications(AbstractNotifications):
    def __init__(self, smtp_host=DEFAULT_HOST, port=DEFAULT_PORT):
        self.server = smtplib.SMTP(smtp_host, port=port)
        self.server.noop()

    def send(self, destination, message):
        msg = f"Subject: allocation service notification\n{message}"
        self.server.sendmail(
            from_addr="allocations@example.com",
            to_addrs=[destination],
            msg=msg,
        )

We change the dependency in the bootstrap script:

我们在引导脚本中更改依赖项：

Example 21. Notifications in message bus (src/allocation/bootstrap.py)（消息总线中的通知）

 def bootstrap(
     start_orm: bool = True,
     uow: unit_of_work.AbstractUnitOfWork = unit_of_work.SqlAlchemyUnitOfWork(),
-    send_mail: Callable = email.send,
+    notifications: AbstractNotifications = EmailNotifications(),
     publish: Callable = redis_eventpublisher.publish,
 ) -> messagebus.MessageBus:

Make a Fake Version for Your Tests

为你的测试创建一个伪造版本

We work through and define a fake version for unit testing:

我们逐步完成并定义一个用于单元测试的伪版本：

Example 22. Fake notifications (tests/unit/test_handlers.py)（伪造通知）

class FakeNotifications(notifications.AbstractNotifications):
    def __init__(self):
        self.sent = defaultdict(list)  # type: Dict[str, List[str]]

    def send(self, destination, message):
        self.sent[destination].append(message)
...

And we use it in our tests:

然后我们在测试中使用它：

Example 23. Tests change slightly (tests/unit/test_handlers.py)（测试略有变化）

    def test_sends_email_on_out_of_stock_error(self):
        fake_notifs = FakeNotifications()
        bus = bootstrap.bootstrap(
            start_orm=False,
            uow=FakeUnitOfWork(),
            notifications=fake_notifs,
            publish=lambda *args: None,
        )
        bus.handle(commands.CreateBatch("b1", "POPULAR-CURTAINS", 9, None))
        bus.handle(commands.Allocate("o1", "POPULAR-CURTAINS", 10))
        assert fake_notifs.sent["stock@made.com"] == [
            f"Out of stock for POPULAR-CURTAINS",
        ]

Figure Out How to Integration Test the Real Thing

找出如何对真实实现进行集成测试

Now we test the real thing, usually with an end-to-end or integration test. We’ve used MailHog as a real-ish email server for our Docker dev environment:

现在我们来测试真实的实现，通常使用端到端或集成测试。我们曾在 Docker 开发环境中使用过 MailHog 作为一个接近真实的邮件服务器：

Example 24. Docker-compose config with real fake email server (docker-compose.yml)（使用真实伪造邮件服务器的 Docker-compose 配置）

version: "3"

services:

  redis_pubsub:
    build:
      context: .
      dockerfile: Dockerfile
    image: allocation-image
    ...

  api:
    image: allocation-image
    ...

  postgres:
    image: postgres:9.6
    ...

  redis:
    image: redis:alpine
    ...

  mailhog:
    image: mailhog/mailhog
    ports:
      - "11025:1025"
      - "18025:8025"

In our integration tests, we use the real EmailNotifications class, talking to the MailHog server in the Docker cluster:

在我们的集成测试中，我们使用真实的 EmailNotifications 类，与 Docker 集群中的 MailHog 服务器通信：

Example 25. Integration test for email (tests/integration/test_email.py)（电子邮件的集成测试）

@pytest.fixture
def bus(sqlite_session_factory):
    bus = bootstrap.bootstrap(
        start_orm=True,
        uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory),
        notifications=notifications.EmailNotifications(),  #(1)
        publish=lambda *args: None,
    )
    yield bus
    clear_mappers()


def get_email_from_mailhog(sku):  #(2)
    host, port = map(config.get_email_host_and_port().get, ["host", "http_port"])
    all_emails = requests.get(f"http://{host}:{port}/api/v2/messages").json()
    return next(m for m in all_emails["items"] if sku in str(m))


def test_out_of_stock_email(bus):
    sku = random_sku()
    bus.handle(commands.CreateBatch("batch1", sku, 9, None))  #(3)
    bus.handle(commands.Allocate("order1", sku, 10))
    email = get_email_from_mailhog(sku)
    assert email["Raw"]["From"] == "allocations@example.com"  #(4)
    assert email["Raw"]["To"] == ["stock@made.com"]
    assert f"Out of stock for {sku}" in email["Raw"]["Data"]

1. We use our bootstrapper to build a message bus that talks to the real notifications class. 我们使用引导程序构建一个使用真实通知类的消息总线。

2. We figure out how to fetch emails from our "real" email server. 我们找出如何从我们的“真实”邮件服务器中获取邮件。

3. We use the bus to do our test setup. 我们使用消息总线来进行测试设置。

4. Against all the odds, this actually worked, pretty much at the first go! 出乎意料的是，这实际上差不多一次就成功了！

And that’s it really.

事情就是这样。

Exercise for the Reader 2（.读者练习 2）

You could do two things for practice regarding adapters:

关于适配器，你可以通过以下两件事来进行练习：

1. Try swapping out our notifications from email to SMS notifications using Twilio, for example, or Slack notifications. Can you find a good equivalent to MailHog for integration testing? 尝试将我们的通知从电子邮件切换为使用 Twilio 的短信通知，或者切换为 Slack 通知。你能找到一个适合集成测试的、类似 MailHog 的工具吗？

2. In a similar way to what we did moving from send_mail to a Notifications class, try refactoring our redis_eventpublisher that is currently just a Callable to some sort of more formal adapter/base class/protocol. 类似我们从 send_mail 转换为 Notifications 类的过程，尝试将目前只是一个 Callable 的 redis_eventpublisher 重构为 某种更正式的适配器/基类/协议。

Wrap-Up

总结

• Once you have more than one adapter, you’ll start to feel a lot of pain from passing dependencies around manually, unless you do some kind of dependency injection. 一旦你有了多个适配器，如果不使用某种 依赖注入，你会在手动传递依赖时感受到很多痛苦。

• Setting up dependency injection is just one of many typical setup/initialization activities that you need to do just once when starting your app. Putting this all together into a bootstrap script is often a good idea. 设置依赖注入只是启动应用程序时只需执行一次的许多典型设置/初始化活动之一。将所有这些整合到一个 引导脚本 中通常是个不错的主意。

• The bootstrap script is also good as a place to provide sensible default configuration for your adapters, and as a single place to override those adapters with fakes for your tests. 引导脚本还是一个为适配器提供合理默认配置的好地方，同时也是统一用伪实现替换这些适配器以便进行测试的地方。

• A dependency injection framework can be useful if you find yourself needing to do DI at multiple levels—if you have chained dependencies of components that all need DI, for example. 如果你发现需要在多个层级上进行依赖注入（DI）——例如如果你有需要 DI 的组件依赖链——那么使用一个依赖注入框架可能会很有用。

• This chapter also presented a worked example of changing an implicit/simple dependency into a "proper" adapter, factoring out an ABC, defining its real and fake implementations, and thinking through integration testing. 本章还展示了一个将隐式/简单依赖转变为“正式”适配器的完整示例，提取了一个抽象基类（ABC），定义了其真实和伪实现，并深入思考了集成测试过程。

DI and Bootstrap Recap（依赖注入与引导函数回顾）

In summary:

总结：

1. Define your API using an ABC. 使用抽象基类（ABC）定义你的 API。

2. Implement the real thing. 实现真实的功能。

3. Build a fake and use it for unit/service-layer/handler tests. 构建一个伪实现，并在单元测试/服务层测试/处理器测试中使用它。

4. Find a less fake version you can put into your Docker environment. 找到一个可以放入你的 Docker 环境中的更接近真实的版本。

5. Test the less fake "real" thing. 测试这个更接近真实的“伪真实”版本。

6. Profit! 获益！

These were the last patterns we wanted to cover, which brings us to the end of [part2]. In the epilogue, we’ll try to give you some pointers for applying these techniques in the Real World^TM.

这些是我们想要涵盖的最后几个模式，这也将我们带到了 [part2] 的结尾。在 尾声 中， 我们将尝试为你提供一些建议，帮助你在真实世界^TM中应用这些技术。

---

1. Because Python is not a "pure" OO language, Python developers aren’t necessarily used to the concept of needing to compose a set of objects into a working application. We just pick our entrypoint and run code from top to bottom.

2. Mark Seemann calls this Pure DI or sometimes Vanilla DI.

3. However, it’s still a global in the flask_app module scope, if that makes sense. This may cause problems if you ever find yourself wanting to test your Flask app in-process by using the Flask Test Client instead of using Docker as we do. It’s worth researching Flask app factories if you get into this.