如何用Python方式进行依赖注入?

16 投票
5 回答
11731 浏览
提问于 2025-04-15 22:04

最近我一直在研究Python的用法,所以我想问一下:

在Python中,怎么做依赖注入呢?

我说的是一些常见的情况,比如服务A需要访问用户服务(UserService)来进行授权检查。

5 个回答

2

什么是依赖注入?

依赖注入是一种原则,帮助减少组件之间的紧耦合,同时增加它们的内聚性。

紧耦合和内聚性是用来描述组件之间关系的。

  • 高耦合。如果耦合度高,就像用超级胶水或者焊接一样,拆卸起来非常困难。
  • 高内聚。高内聚就像用螺丝钉,拆卸和重新组装都很简单,甚至可以以不同的方式组装。这与高耦合是相反的。

当内聚性高时,耦合度就低。

低耦合带来了灵活性。你的代码更容易修改和测试。

如何实现依赖注入?

对象之间不再互相创建。它们提供了一种方式来注入依赖。

之前:


import os


class ApiClient:

    def __init__(self):
        self.api_key = os.getenv('API_KEY')  # <-- dependency
        self.timeout = os.getenv('TIMEOUT')  # <-- dependency


class Service:

    def __init__(self):
        self.api_client = ApiClient()  # <-- dependency


def main() -> None:
    service = Service()  # <-- dependency
    ...


if __name__ == '__main__':
    main()

之后:


import os


class ApiClient:

    def __init__(self, api_key: str, timeout: int):
        self.api_key = api_key  # <-- dependency is injected
        self.timeout = timeout  # <-- dependency is injected


class Service:

    def __init__(self, api_client: ApiClient):
        self.api_client = api_client  # <-- dependency is injected


def main(service: Service):  # <-- dependency is injected
    ...


if __name__ == '__main__':
    main(
        service=Service(
            api_client=ApiClient(
                api_key=os.getenv('API_KEY'),
                timeout=os.getenv('TIMEOUT'),
            ),
        ),
    )

ApiClient 不再需要知道选项来自哪里。你可以从配置文件中读取一个键和超时时间,甚至可以从数据库中获取。

Service 不再依赖于 ApiClient。它不再创建它。你可以提供一个模拟对象或其他兼容的对象。

函数 main() 不再依赖于 Service。它将 Service 作为参数传入。

灵活性是有代价的。

现在你需要像这样组装和注入对象:


main(
    service=Service(
        api_client=ApiClient(
            api_key=os.getenv('API_KEY'),
            timeout=os.getenv('TIMEOUT'),
        ),
    ),
)

组装代码可能会重复,这样会让改变应用结构变得更加困难。

总结

依赖注入带来了三个好处:

  • 灵活性。组件之间的耦合较松。你可以通过不同的方式组合组件,轻松扩展或改变系统的功能,甚至可以实时进行。
  • 可测试性。测试变得简单,因为你可以轻松地注入模拟对象,而不是使用真实的API或数据库对象等。
  • 清晰性和可维护性。依赖注入帮助你明确依赖关系。隐式的变得显式。“显式优于隐式”(PEP 20 - Python之禅)。你在容器中明确定义了所有组件和依赖关系。这提供了对应用结构的概览和控制,容易理解和修改。

——

我相信通过已经展示的例子,你会理解这个概念,并能够将其应用到你的问题上,比如实现 UserService 用于授权

2

经过多年的使用,我发现用Python写简单代码时,其实不需要像Java的Spring那样的依赖注入框架。依赖注入就是把一个对象的依赖关系(比如需要用到的其他对象)传递给它,而不是让它自己去找。比如,像下面这样做就可以了:

def foo(dep = None):  # great for unit testing!
    self.dep = dep or Dep()  # callers can not care about this too
    ...

这就是纯粹的依赖注入,方法很简单,但没有使用那些自动帮你注入的神奇框架(也就是所谓的自动注入),也没有控制反转的概念。

不过,当我处理更大的应用时,这种方法就不够用了。所以我想出了一个叫做 injectable 的微框架,它不会让你觉得不符合Python的风格,同时又能提供一流的依赖注入自动注入功能。

我们的口号是 为人类提供依赖注入™,它的样子是这样的:

# some_service.py
class SomeService:
    @autowired
    def __init__(
        self,
        database: Autowired(Database),
        message_brokers: Autowired(List[Broker]),
    ):
        pending = database.retrieve_pending_messages()
        for broker in message_brokers:
            broker.send_pending(pending)
# database.py
@injectable
class Database:
    ...
# message_broker.py
class MessageBroker(ABC):
    def send_pending(messages):
        ...
# kafka_producer.py
@injectable
class KafkaProducer(MessageBroker):
    ...
# sqs_producer.py
@injectable
class SQSProducer(MessageBroker):
    ...
15

这完全要看具体情况。比如说,如果你为了测试的目的使用依赖注入——这样你就可以轻松地模拟某些东西——那么你有时可以完全不使用注入:你可以直接模拟你本来会注入的模块或类。

subprocess.Popen = some_mock_Popen
result = subprocess.call(...)
assert some_mock_popen.result == result

subprocess.call() 会调用 subprocess.Popen(),而我们可以直接模拟它,而不需要以特殊的方式注入依赖。我们可以直接替换 subprocess.Popen。 (这只是一个例子;在实际情况中,你会以更稳健的方式来处理。)

如果你在更复杂的情况下使用依赖注入,或者当模拟整个模块或类不合适时(比如说,你只想模拟某个特定的调用),那么通常会选择使用类属性或模块全局变量来处理依赖。举个例子,考虑一个 my_subprocess.py

from subprocess import Popen

def my_call(...):
    return Popen(...).communicate()

你可以通过给 my_subprocess.Popen 赋值,轻松替换 my_call() 中的 Popen 调用;这不会影响对 subprocess.Popen 的其他调用(当然,它会替换所有对 my_subprocess.Popen 的调用)。同样,类属性也是如此:

class MyClass(object):
    Popen = staticmethod(subprocess.Popen)
    def call(self):
        return self.Popen(...).communicate(...)

使用类属性时,虽然考虑到其他选项,这种情况很少需要,但你应该小心使用 staticmethod。如果不这样做,而你插入的对象是一个普通的函数对象或其他类型的描述符,比如属性,当从类或实例中获取时会做一些特殊的事情,这样就会出错。更糟的是,如果你使用的东西现在不是描述符(比如在这个例子中的 subprocess.Popen 类),现在可以正常工作,但如果将来这个对象变成了普通函数,就会让人困惑地出错。

最后,还有简单的回调;如果你只是想把某个类的特定实例与某个特定服务关联起来,你可以直接把服务(或服务的一个或多个方法)传递给类的初始化器,然后让它使用这些服务:

class MyClass(object):
    def __init__(self, authenticate=None, authorize=None):
        if authenticate is None:
            authenticate = default_authenticate
        if authorize is None:
            authorize = default_authorize
        self.authenticate = authenticate
        self.authorize = authorize
    def request(self, user, password, action):
        self.authenticate(user, password)
        self.authorize(user, action)
        self._do_request(action)

...
helper = AuthService(...)
# Pass bound methods to helper.authenticate and helper.authorize to MyClass.
inst = MyClass(authenticate=helper.authenticate, authorize=helper.authorize)
inst.request(...)

在这样设置实例属性时,你不需要担心描述符的触发,所以直接赋值给函数(或类或其他可调用对象或实例)就可以了。

撰写回答