rodi: Python 3 的依赖注入实现

rodi: Python 3 的依赖注入实现#

功能：

  • 类型解析(types resolution)通过签名类型注解（类型提示）

  • 类型解析通过类注解（类型提示）

  • 类型解析通过名称和别名（约定优于配置）

  • 无侵入式：无需更改类源代码即可构建对象图

  • 最低开销以获取服务，一旦对象图构建完成

  • 支持单例、瞬态和作用域服务

rodi 在运行时检查代码一次，生成返回所需类型实例的函数 - 只要对象图未被更改。检查是在构造函数（__init__）或类注解上进行的。验证步骤，例如检测循环依赖或缺失的服务，是在构建这些函数时进行的，因此激活服务时不需要额外的验证。

rodi 提供两个代码 API：

  • 一种尽可能保持通用性，使用 ContainerProtocol 来表示那些希望能够用 Python 的依赖注入的替代实现替换 rodi 的场景。该协议只期望一个能够 register 和 resolve 类型，并能够判断是否在其中配置了类型（ __contains__ ）。即使其他依赖注入的实现没有实现这三个方法，也应该很容易使用组合来包装其他库的兼容类。

  • 一种是一个更具体的实现，对于不希望考虑依赖注入的替代实现的情况。

推荐做法

所有服务应在应用程序启动时配置一次，并且在正常程序执行期间不应更改对象图。例如：如果您构建一个 Web 应用程序，则在启动应用程序时配置对象图，处理 Web 请求时避免更改 Container 配置。

旨在将 Container 和服务图从应用程序的前端层抽象出来，并避免将运行时值与容器配置混合。例如：如果您构建 Web 应用程序，尽可能避免依赖于 HTTP 请求对象是您容器中注册的服务。

Service life style

  • 单例(singleton) - 每个服务提供商仅实例化一次

  • 瞬态(transient) - 服务每次需要时都会被实例化

  • 作用域(作用域) - 仅在每个根服务解析调用中实例化一次（例如，每个 Web 请求一次）

示例1#

本例说明了使用 Container 类注册两种类型的基本用法，并通过类型检查实现自动解析。 两个服务注册为“瞬态”服务，意味着每次需要时都会创建新的实例。

class A:
    ...


class B:
    friend: A

from rodi import Container

container = Container()

container.register(A)
container.register(B)
example_1 = container.resolve(B)
assert isinstance(example_1, B)
assert isinstance(example_1.friend, A)

example_2 = container.resolve(B)
assert isinstance(example_2, B)
assert isinstance(example_2.friend, A)

assert example_1 is not example_2

---------------------------------------------------------------------------
ModuleNotFoundError                       Traceback (most recent call last)
Cell In[2], line 1
----> 1 from rodi import Container
      3 container = Container()
      5 container.register(A)

ModuleNotFoundError: No module named 'rodi'

示例2#

本例说明了通过基类型注册具体类型并按基类型激活 Container 的基本用法。

这种模式有助于编写解耦的代码（例如，业务层逻辑与数据访问逻辑的具体实现分离）。

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class Cat:
    id: str
    name: str


class CatsRepository(ABC):
    @abstractmethod
    def get_cat(self, cat_id: str) -> Cat:
        """Gets information of a cat by ID."""


class SQLiteCatsRepository(CatsRepository):
    def get_cat(self, cat_id: str) -> Cat:
        """Gets information of a cat by ID, from a source SQLite DB."""
        raise NotImplementedError()

from rodi import Container

container = Container()

container.register(CatsRepository, SQLiteCatsRepository)

example_1 = container.resolve(CatsRepository)

assert isinstance(example_1, SQLiteCatsRepository)

示例3#

本例说明如何配置单例对象。

from dataclasses import dataclass

@dataclass
class Cat:
    id: str
    name: str

使用 ContainerProtocol（建议在可能需要用其他依赖注入实现替换当前库时采用）：

from rodi import Container

container = Container()
container.register(Cat, instance=Cat("1", "Celine"))
example = container.resolve(Cat)

assert isinstance(example, Cat)
assert example.id == "1" and example.name == "Celine"

assert example is container.resolve(Cat)

使用原始代码的 API：

class Foo:
    ...


container.add_instance(Foo())

assert container.resolve(Foo) is container.resolve(Foo)

更多教程见：dependency-injection

类型 A 被注册为瞬态， B 为作用域， C 为单例：

from dataclasses import dataclass
class A:
    ...


class B:
    ...


class C:
    ...


@dataclass
class Foo:
    a1: A
    a2: A
    b1: B
    b2: B
    c1: C
    c2: C

from rodi import Container

services = Container()
services.add_transient(A)
services.add_scoped(B)
services.add_singleton(C)
services.add_scoped(Foo)

def test(foo: Foo):
    return f"""
    A1: {id(foo.a1)}

    A2: {id(foo.a2)}

    B1: {id(foo.b1)}

    B2: {id(foo.b2)}

    C1: {id(foo.c1)}

    C2: {id(foo.c2)}
    """

  • 瞬态服务始终在激活时实例化（A）

  • scoped services 在每个 Web 请求中只实例化一次（B）

  • 单例服务仅激活一次（C）

目录