如何用Python方式进行依赖注入?
最近我一直在研究Python的用法,所以我想问一下:
在Python中,怎么做依赖注入呢?
我说的是一些常见的情况,比如服务A需要访问用户服务(UserService)来进行授权检查。
5 个回答
什么是依赖注入?
依赖注入是一种原则,帮助减少组件之间的紧耦合,同时增加它们的内聚性。
紧耦合和内聚性是用来描述组件之间关系的。
- 高耦合。如果耦合度高,就像用超级胶水或者焊接一样,拆卸起来非常困难。
- 高内聚。高内聚就像用螺丝钉,拆卸和重新组装都很简单,甚至可以以不同的方式组装。这与高耦合是相反的。
当内聚性高时,耦合度就低。
低耦合带来了灵活性。你的代码更容易修改和测试。
如何实现依赖注入?
对象之间不再互相创建。它们提供了一种方式来注入依赖。
之前:
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 用于授权。
经过多年的使用,我发现用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):
...
这完全要看具体情况。比如说,如果你为了测试的目的使用依赖注入——这样你就可以轻松地模拟某些东西——那么你有时可以完全不使用注入:你可以直接模拟你本来会注入的模块或类。
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(...)
在这样设置实例属性时,你不需要担心描述符的触发,所以直接赋值给函数(或类或其他可调用对象或实例)就可以了。