Python的依赖注入（DI）：如何实现一个简单的依赖注入容器，降低模块间的耦合度。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Python 依赖注入 (DI): 构建解耦的应用程序

大家好！今天我们来聊聊Python中的依赖注入（Dependency Injection, DI）。在构建大型、可维护的应用程序时，模块间的耦合度是一个至关重要的问题。高耦合度意味着一个模块的修改可能会导致其他模块产生意想不到的变化，从而使代码难以理解、测试和重用。依赖注入作为一种设计模式，能够有效地降低模块间的耦合度，提高代码的灵活性和可测试性。

什么是依赖注入？

简单来说，依赖注入是一种设计模式，它将对象的依赖关系从对象本身转移到外部容器或框架来管理。这意味着对象不再负责创建或查找它们所依赖的其他对象，而是由外部“注入”这些依赖项。

更具体地说，依赖注入包含三个关键角色：

服务 (Service): 提供某种功能的组件。
客户 (Client): 需要使用服务的组件。
注入器 (Injector/Container): 负责将服务注入到客户。

想象一下你想要喝咖啡。

服务: 咖啡机 (提供咖啡)
客户: 你 (需要咖啡)
注入器: 咖啡店服务员 (把咖啡递给你)

你不需要自己去制造咖啡机，也不需要自己去寻找咖啡豆。咖啡店服务员负责把咖啡递给你，这就是依赖注入的本质。

依赖注入的好处

降低耦合度: 客户类不再依赖于服务的具体实现，而是依赖于抽象接口。这使得你可以轻松地替换不同的服务实现，而无需修改客户类的代码。
提高可测试性: 通过依赖注入，你可以很容易地使用 Mock 对象来模拟服务的行为，从而对客户类进行单元测试。
提高可重用性: 客户类不再与特定的服务实现绑定，因此可以更容易地在不同的上下文中重用。
提高可维护性: 由于模块间的耦合度降低，代码更加模块化，更容易理解和维护。
提高灵活性: 可以动态地配置和切换不同的服务实现，从而提高应用程序的灵活性。

依赖注入的类型

依赖注入主要有三种类型：

类型	描述	示例
构造器注入	通过类的构造函数来注入依赖项。	`python class UserService: def __init__(self, user_repository): self.user_repository = user_repository def get_user(self, user_id): return self.user_repository.get_user(user_id)`
属性注入 (Setter注入)	通过类的属性（通常是 setter 方法）来注入依赖项。	`python class UserService: def __init__(self): self.user_repository = None def set_user_repository(self, user_repository): self.user_repository = user_repository def get_user(self, user_id): return self.user_repository.get_user(user_id)`
接口注入	通过接口定义一个注入方法，客户类实现该接口并接收依赖项。这种方式在Python中相对少见，因为Python的动态类型特性使得通常不需要显式接口。但如果需要强制执行某种注入方式，接口注入可以提供更强的约束。	`python from abc import ABC, abstractmethod class UserRepositoryInterface(ABC): @abstractmethod def get_user(self, user_id): pass class ServiceInjectorInterface(ABC): @abstractmethod def inject_user_repository(self, repository): pass class UserService(ServiceInjectorInterface): def __init__(self): self.user_repository = None def inject_user_repository(self, user_repository): self.user_repository = user_repository def get_user(self, user_id): return self.user_repository.get_user(user_id)`

一般来说，构造器注入是首选的注入方式，因为它能够确保客户类在创建时就拥有所需的依赖项，并使依赖关系更加明确。

手动实现一个简单的依赖注入容器

现在，让我们来手动实现一个简单的依赖注入容器，以便更好地理解依赖注入的原理。

class Container:
    def __init__(self):
        self.services = {}

    def register(self, name, provider):
        """
        注册一个服务。

        Args:
            name: 服务的名称（字符串）。
            provider: 一个函数，用于创建服务的实例。
        """
        self.services[name] = provider

    def resolve(self, name):
        """
        解析一个服务。

        Args:
            name: 服务的名称（字符串）。

        Returns:
            服务的实例。
        """
        if name not in self.services:
            raise Exception(f"Service '{name}' not registered.")
        provider = self.services[name]
        return provider()

# 示例服务
class UserRepository:
    def get_user(self, user_id):
        # 模拟从数据库获取用户
        return f"User from DB with ID: {user_id}"

class UserService:
    def __init__(self, user_repository):
        self.user_repository = user_repository

    def get_user(self, user_id):
        return self.user_repository.get_user(user_id)

# 创建容器
container = Container()

# 注册服务
container.register("user_repository", UserRepository)  #注意这里传递的是类名，而不是类的实例。
container.register("user_service", lambda: UserService(container.resolve("user_repository"))) #这里使用了lambda表达式，延迟了user_repository的创建，直到UserService被解析时才创建。

# 解析服务
user_service = container.resolve("user_service")

# 使用服务
user = user_service.get_user(123)
print(user)  # 输出: User from DB with ID: 123

在这个例子中，Container 类充当了依赖注入容器的角色。register 方法用于注册服务，resolve 方法用于解析服务。UserService 依赖于 UserRepository，但它并不负责创建 UserRepository 的实例，而是由容器来注入。

代码解释:

Container 类:
- __init__: 初始化一个空的 services 字典，用于存储已注册的服务。
- register(self, name, provider): 注册一个服务。name 是服务的名称，provider 是一个函数，当需要创建服务实例时，会调用这个函数。关键点在于，provider 是一个函数，而不是一个实例。这允许我们延迟创建服务实例，直到真正需要它的时候。
- resolve(self, name): 解析一个服务。它首先检查服务是否已注册，然后调用 provider 函数来创建服务的实例。
UserRepository 类:
- 一个简单的类，模拟从数据库获取用户。
UserService 类:
- 依赖于 UserRepository。在构造函数中，它接收一个 user_repository 实例。
- get_user(self, user_id) 方法调用 user_repository 的 get_user 方法来获取用户。
容器的使用:
- 创建 Container 的实例。
- 使用 container.register 注册 user_repository 和 user_service。注意，user_service 的 provider 是一个 lambda 函数，它接受一个 container 参数，并使用 container.resolve 来解析 user_repository。
- 使用 container.resolve 解析 user_service。
- 调用 user_service 的 get_user 方法来获取用户。

为什么要使用 lambda 表达式？

在注册 user_service 时，我们使用了 lambda 表达式：

container.register("user_service", lambda: UserService(container.resolve("user_repository")))

这是因为我们希望延迟创建 UserService 实例，直到真正需要它的时候。如果我们直接传递 UserService(container.resolve("user_repository"))，那么 UserService 实例会在注册时立即创建，这可能不是我们想要的。使用 lambda 表达式，我们将创建 UserService 实例的逻辑封装在一个函数中，只有在调用 container.resolve("user_service") 时，这个函数才会被执行，从而创建 UserService 实例。

这个简单的依赖注入容器展示了依赖注入的核心思想：将对象的依赖关系从对象本身转移到外部容器来管理。

进一步改进依赖注入容器

上面的示例只是一个非常简单的依赖注入容器。我们可以对其进行一些改进，以使其更加强大和灵活。

1. 自动依赖解析:

我们可以让容器自动解析构造函数的参数，而不需要手动提供 provider 函数。

import inspect

class Container:
    def __init__(self):
        self.services = {}

    def register(self, name, cls):
        """
        注册一个服务。

        Args:
            name: 服务的名称（字符串）。
            cls: 服务的类。
        """
        self.services[name] = cls

    def resolve(self, name):
        """
        解析一个服务。

        Args:
            name: 服务的名称（字符串）。

        Returns:
            服务的实例。
        """
        if name not in self.services:
            raise Exception(f"Service '{name}' not registered.")
        cls = self.services[name]
        return self._build(cls)

    def _build(self, cls):
        """
        自动构建类的实例，并解析其依赖项。

        Args:
            cls: 要构建的类。

        Returns:
            类的实例。
        """
        sig = inspect.signature(cls.__init__)
        params = []
        for param in sig.parameters.values():
            if param.name == 'self':
                continue
            if param.annotation == inspect.Parameter.empty:
                raise Exception(f"Dependency '{param.name}' for class '{cls.__name__}' has no type hint.")
            dependency = self.resolve(param.annotation.__name__.lower()) #这里假设依赖项的名称是类名的小写形式
            params.append(dependency)

        return cls(*params)

# 示例服务
class UserRepository:
    def get_user(self, user_id):
        # 模拟从数据库获取用户
        return f"User from DB with ID: {user_id}"

class UserService:
    def __init__(self, user_repository: UserRepository): # 使用类型提示
        self.user_repository = user_repository

    def get_user(self, user_id):
        return self.user_repository.get_user(user_id)

# 创建容器
container = Container()

# 注册服务 (现在直接注册类)
container.register("user_repository", UserRepository)
container.register("user_service", UserService)

# 解析服务
user_service = container.resolve("user_service")

# 使用服务
user = user_service.get_user(123)
print(user)

在这个改进后的版本中，我们使用 inspect 模块来获取类的构造函数的签名，并自动解析其依赖项。

代码解释:

_build(self, cls) 方法:
- 使用 inspect.signature 获取类的构造函数的签名。
- 遍历构造函数的参数，并解析每个参数的依赖项。
- 使用 self.resolve 来解析依赖项。
- 使用解析后的依赖项来创建类的实例。
类型提示:
- UserService 的构造函数使用了类型提示：user_repository: UserRepository。这使得容器能够自动解析 user_repository 的依赖项。
简化注册:
- 现在，我们可以直接注册类，而不需要提供 provider 函数。

2. 单例模式:

我们可以让容器管理服务的生命周期，例如，只创建一个服务的实例（单例模式）。

class Container:
    def __init__(self):
        self.services = {}
        self.instances = {}

    def register(self, name, cls, singleton=False):
        """
        注册一个服务。

        Args:
            name: 服务的名称（字符串）。
            cls: 服务的类。
            singleton: 是否为单例模式（布尔值）。
        """
        self.services[name] = (cls, singleton)

    def resolve(self, name):
        """
        解析一个服务。

        Args:
            name: 服务的名称（字符串）。

        Returns:
            服务的实例。
        """
        if name not in self.services:
            raise Exception(f"Service '{name}' not registered.")
        cls, singleton = self.services[name]

        if singleton:
            if name not in self.instances:
                self.instances[name] = self._build(cls)
            return self.instances[name]
        else:
            return self._build(cls)

    def _build(self, cls):
        # (省略)  与之前的自动依赖解析代码相同
        ...

# 示例服务
class UserRepository:
    def get_user(self, user_id):
        # 模拟从数据库获取用户
        return f"User from DB with ID: {user_id}"

class UserService:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository

    def get_user(self, user_id):
        return self.user_repository.get_user(user_id)

# 创建容器
container = Container()

# 注册服务 (指定单例模式)
container.register("user_repository", UserRepository, singleton=True)
container.register("user_service", UserService)

# 解析服务
user_service1 = container.resolve("user_service")
user_service2 = container.resolve("user_service")

# 检查 user_repository 是否为单例
print(user_service1.user_repository is user_service2.user_repository) # 输出: True

在这个版本中，我们添加了一个 singleton 参数到 register 方法，用于指定服务是否为单例模式。如果服务是单例模式，那么容器只会创建一个实例，并在后续的 resolve 调用中返回同一个实例。

代码解释:

singleton 参数:
- register 方法现在接受一个 singleton 参数，用于指定服务是否为单例模式。
instances 字典:
- Container 类现在有一个 instances 字典，用于存储单例服务的实例。
resolve 方法:
- 在 resolve 方法中，我们首先检查服务是否为单例模式。如果是，我们检查 instances 字典中是否已经存在该服务的实例。如果存在，我们直接返回该实例。否则，我们创建一个新的实例，并将其存储在 instances 字典中。

使用第三方依赖注入库

虽然我们可以手动实现一个简单的依赖注入容器，但在实际项目中，通常建议使用第三方依赖注入库。这些库提供了更丰富的功能和更好的性能。

一些流行的 Python 依赖注入库包括：

injector: 一个轻量级的依赖注入框架。
dependency_injector: 一个功能更强大的依赖注入框架，支持多种注入方式和生命周期管理。
pinject: 由 Google 开发的依赖注入框架。

使用第三方库可以简化代码，提高开发效率，并获得更好的可维护性。

例如，使用 dependency_injector：

from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide

class UserRepository:
    def get_user(self, user_id):
        return f"User from DB with ID: {user_id}"

class UserService:
    @inject
    def __init__(self, user_repository: UserRepository = Provide["user_repository"]):
        self.user_repository = user_repository

    def get_user(self, user_id):
        return self.user_repository.get_user(user_id)

class Container(containers.DeclarativeContainer):
    user_repository = providers.Singleton(UserRepository)
    user_service = providers.Factory(UserService)

container = Container()
container.wire(modules=[__name__]) # 需要显式地将容器连接到模块

user_service = container.user_service()
user = user_service.get_user(123)
print(user)

这个例子使用了 dependency_injector 库来实现依赖注入。

代码解释:

containers.DeclarativeContainer:
- 定义一个容器类，用于声明服务。
providers.Singleton 和 providers.Factory:
- providers.Singleton 用于声明单例服务。
- providers.Factory 用于声明每次都创建新实例的服务。
@inject 装饰器:
- 用于标记需要注入依赖项的类的构造函数。
Provide["user_repository"]:
- 用于指定要注入的依赖项的名称。
container.wire(modules=[__name__]):
- 将容器连接到模块，以便 dependency_injector 能够找到需要注入依赖项的类。

如何选择合适的依赖注入方式

选择哪种依赖注入方式取决于项目的具体需求和复杂程度。

手动依赖注入:
- 适用于小型项目或原型项目。
- 简单易懂，但代码量较大，可维护性较差。
第三方依赖注入库:
- 适用于中大型项目。
- 功能强大，性能良好，可维护性高。

在选择第三方库时，需要考虑以下因素：

功能: 库是否提供所需的功能，例如，多种注入方式、生命周期管理、自动依赖解析等。
性能: 库的性能是否满足项目的需求。
易用性: 库是否易于使用和学习。
社区支持: 库是否有活跃的社区支持。

依赖注入的陷阱

虽然依赖注入是一种强大的设计模式，但如果不正确地使用，也可能导致一些问题。

过度使用: 不要过度使用依赖注入。对于简单的类，手动创建依赖项可能更简单。
循环依赖: 避免循环依赖。循环依赖会导致容器无法解析依赖项。
隐藏依赖: 确保依赖关系清晰可见。不要隐藏依赖关系，例如，使用全局变量或单例模式来传递依赖项。

总结：解耦，测试，灵活

今天我们讨论了Python中的依赖注入，包括它的定义、好处、类型以及如何手动实现一个简单的依赖注入容器。依赖注入通过将依赖关系的控制权转移到外部容器，降低模块间的耦合度，提高代码的可测试性和可维护性，并最终构建出更加灵活和健壮的应用程序。在实际项目中，建议使用第三方依赖注入库，以简化代码并提高开发效率。掌握依赖注入可以帮助你编写出更优雅、更易于维护的代码。

Python 依赖注入 (DI): 构建解耦的应用程序

什么是依赖注入？

依赖注入的好处

依赖注入的类型

手动实现一个简单的依赖注入容器

进一步改进依赖注入容器

使用第三方依赖注入库

如何选择合适的依赖注入方式

依赖注入的陷阱

总结：解耦，测试，灵活

发表回复 取消回复

发表回复取消回复