课程
依赖注入是一种设计模式,它将对象或服务(即依赖)从外部提供给类,而不是由类在内部自行创建。这样可以解耦组件。
在 Python 中使用依赖注入有许多好处,例如:
- 模块化: 将代码拆分为更小、可复用的部分。
- 可测试性: 更易于测试,因为您可以用模拟对象替换真实部件。
- 可维护性: 可以更新或修改代码的部分而不破坏系统的其余部分。
在本教程中,我将通过简单、实用的示例演示如何在 Python 中使用依赖注入。
要点速览
-
依赖注入是一种设计模式,类从外部接收其依赖,而不是自己创建它们。它让您的代码更加模块化。
-
您可以通过构造函数、setter 方法或直接在方法中注入依赖;Python 可手动实现,也可借助
dependency-injector、injector和punq等框架。 -
常见 Web 框架处理方式不同:FastAPI 通过
Depends()内置支持。Flask 和 Django 需要配合dependency-injector等库。
理解 Python 中的依赖注入
要实现依赖注入,首先需要掌握其背后的关键原则。
依赖注入基于控制反转(IoC)原则:类不再自己创建和管理依赖,而是从外部接收它们。这有助于聚焦单一职责,让代码更干净。
在 Python 中注入依赖常见方式包括(还有其他方法):
- 构造函数注入: 通过类的构造函数传入依赖。
- Setter 注入: 对象创建后使用 setter 方法设置依赖。
- 方法注入: 将依赖直接传递给需要它们的方法。
为什么在 Python 中使用依赖注入?
作为开发者,理解并实现依赖注入会显著改善您的代码设计。我们来看一些依赖注入的好处。
模块化、可复用性与灵活性
依赖注入使您可以将代码拆分为更小、更专注的组件,从而获得更高的灵活性和模块化。
每个部分或组件只处理特定任务,并依赖外部依赖,因此更易在您或他人的应用中复用。
例如,若您有一个发送消息的类。与其把邮件或短信逻辑直接写进服务,不如注入一个邮件或短信发送器类。
这样,您的消息发送类无需关心消息如何发送,只需使用注入的发送器即可。这意味着您可以与不同的邮件或短信发送器搭配使用。
可维护性与可测试性
使用依赖注入时,您可以在测试中轻松用模拟数据替换真实依赖。这样便于编写无需依赖 API 或数据库的单元测试。
设想一下,您有一个将数据保存到数据库的类。出于测试目的,您可以注入一个模拟数据库而非真实数据库,避免测试与生产数据库耦合。
依赖注入让您可以修改代码的特定部分而不破坏整个系统。尤其在需要更换应用中的支付服务商时(如从 Stripe 切换到 PayPal),无需修改其余代码。
控制反转与可替换的依赖
控制反转的核心思想是:依赖注入把创建和管理依赖的职责转移到外部来源。
这意味着您可以轻松地用另一种实现替换现有实现;例如,您有一个 ReportGenerator 类用于格式化并导出报告。
与其由类来决定导出格式,不如向其注入一个导出器类,例如 PDFExporter、ExcelExporter 或 CSVExporter。
ReportGenerator 类并不关心导出如何进行;要更改格式,只需替换注入的导出器,无需修改 ReportGenerator 本身。
在 Python 中实现依赖注入
现在您已清楚依赖注入是什么以及为何有用,接下来看看如何在 Python 中实现它。我们先从最简单的方法开始:手动依赖注入。
手动依赖注入
在手动依赖注入中,您自行创建依赖,并将它们传递给需要它们的类或函数。
该方法适合小型应用,但随着应用扩展,您需要使用依赖注入框架以避免出错。
以下是不使用依赖注入的示例代码
class EmailService:
def send_email(self, message):
print(f"Sending email: {message}")
class UserNotifier:
def __init__(self):
self.email_service = EmailService() # Creates its own dependency
def notify(self, message):
self.email_service.send_email(message)
notifier = UserNotifier()
notifier.notify("Welcome!")
在上述示例中,UserNotifier 与 EmailService 紧密耦合。您无法在测试中轻松替换或模拟 EmailService。
下面是使用手动依赖注入的另一个版本。
class EmailService:
def send_email(self, message):
print(f"Sending email: {message}")
class UserNotifier:
def __init__(self, email_service):
self.email_service = email_service # Dependency is injected
def notify(self, message):
self.email_service.send_email(message)
email_service = EmailService()
notifier = UserNotifier(email_service)
notifier.notify("Welcome!")
上述版本更灵活,允许您注入不同的消息服务;在测试中也可以用模拟对象替换 EmailService。
使用 dependency-injector 框架
随着应用增长,管理依赖会变得复杂而混乱。
dependency-injector 框架提供了一种结构化的方法。
容器-提供者架构
dependency-injector 基于容器-提供者架构运作。
- 容器:应用依赖的中心注册表。
- 提供者:定义如何创建依赖。
- 配置:允许修改设置并将其注入对象。
- 覆盖:让您无需改动应用代码即可替换依赖,尤其适用于测试期间。
from dependency_injector import containers, providers
# Services
class EmailService:
def send_email(self, message):
print(f"Sending email: {message}")
class UserNotifier:
def __init__(self, email_service):
self.email_service = email_service
def notify(self, message):
self.email_service.send_email(message)
# Container
class Container(containers.DeclarativeContainer):
email_service = providers.Singleton(EmailService)
user_notifier = providers.Factory(UserNotifier, email_service=email_service)
# Usage
container = Container()
notifier = container.user_notifier()
notifier.notify("Hello!")
在上述示例中:
-
Container定义了实例的创建方式 -
Singleton确保EmailService只有一个实例 -
Factory创建新的UserNotifier,并自动注入EmailService。
Wiring 机制
您可以使用如 装饰器 @inject 之类的方式来减少手动传递依赖。
下面是上一个示例的另一个版本:
from dependency_injector import containers, providers
# Define services
class EmailService:
def send_email(self, message):
print(f"Sending email: {message}")
class UserNotifier:
def __init__(self, email_service: EmailService):
self.email_service = email_service
def notify(self, message):
self.email_service.send_email(message)
# Create container
class Container(containers.DeclarativeContainer):
email_service = providers.Singleton(EmailService)
user_notifier = providers.Factory(UserNotifier, email_service=email_service)
from dependency_injector.wiring import inject, Provide
@inject
def main(notifier: UserNotifier = Provide[Container.user_notifier]):
notifier.notify("Hello!")
if __name__ == "__main__":
container = Container()
container.wire(modules=[__name__]) # This wires the current module
main()
在当前版本中:
-
无需手动添加依赖,
@inject装饰器会自动注入函数或类的依赖。 -
Provide类告知dependency-injector注入哪一个依赖。
这种方式免除了从容器手动连接依赖的需求。
在更大型的应用中非常有用,因为服务可能依赖多个组件(因此需要自动初始化)。
在主流 Python 框架中使用依赖注入
大多数现代 Python 框架都便于使用依赖注入,使您能编写更整洁、更易测试的代码。
本节将介绍 Flask、Django 和 FastAPI 中依赖注入的用法。
Flask 中的依赖注入
Flask 没有内置依赖注入支持,但您可以通过 dependency-injector 库来实现,步骤如下。
步骤 1:定义服务
class GreetingService:
def get_greeting(self, name):
return f"Hello, {name}!"
步骤 2:设置依赖注入容器
from dependency_injector import containers, providers
class Container(containers.DeclarativeContainer):
greeting_service = providers.Factory(GreetingService)
步骤 3:创建 Flask 应用并注入依赖
from flask import Flask, request
from dependency_injector.wiring import inject, Provide
container = Container()
app = Flask(__name__)
@app.route("/greet")
@inject
def greet(greeting_service: GreetingService = Provide[Container.greeting_service]):
name = request.args.get("name", "World")
return greeting_service.get_greeting(name)
container.wire(modules=[__name__])
if __name__ == "__main__":
app.run(debug=True)
从上述代码可以看到:
-
函数
name = request.args.get(“name”, “World”)从查询字符串中提取name,若未提供则默认为”World”。 -
return greeting_service.get_greeting(name)调用注入服务的方法并返回问候消息。 -
container.wire(modules=[__name__])告诉依赖注入器扫描此模块中的@inject装饰器和Provide[...]标记,并将它们与实际提供者连接。
步骤 4:测试应用
在终端运行命令 python <name_of_file>.py,并访问 URL http://localhost:5000/greet?name=Jacob。
您应当会看到如下内容。

Django 中的依赖注入
与 Flask 类似,Django 也不提供内置依赖注入。但您可以利用其基于类的视图、中间件和应用结构手动集成,或使用 dependency-injector 库。
以下是在 Django 中实现依赖注入的分步流程。
步骤 1:定义服务
在应用文件夹下的新文件 services.py 中创建包含业务逻辑的简单服务。
# myapp/services.py
class GreetingService:
def get_greeting(self, name):
return f"Hello, {name}!"
步骤 2:设置依赖注入容器
在应用目录内,使用 dependency-injector 库在新文件 containers.py 中创建容器。
# myapp/containers.py
from dependency_injector import containers, providers
from .services import GreetingService
class Container(containers.DeclarativeContainer):
greeting_service = providers.Factory(GreetingService)
步骤 3:创建注入依赖的视图
在 views.py 中使用 @inject 装饰器接收依赖。
# myapp/views.py
from django.http import HttpResponse
from dependency_injector.wiring import inject, Provide
from .containers import Container
from .services import GreetingService
@inject
def greet_view(
request, greeting_service: GreetingService = Provide[Container.greeting_service]
):
name = request.GET.get("name", "World")
message = greeting_service.get_greeting(name)
return HttpResponse(message)
步骤 4:更新 urls.py
在项目文件夹下,进入 urls.py 文件,将依赖注入器与 Django 的 URL 路由系统对接。
from django.contrib import admin
from django.urls import path
from myapp.views import greet_view
urlpatterns = [
path("admin/", admin.site.urls),
path("greet/", greet_view),
]
步骤 5:在 apps.py 中连接容器
前往 apps.py,将容器与应用连接。
# myapp/apps.py
from django.apps import AppConfig
from .containers import Container
class MyAppConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "myapp"
def ready(self):
container = Container()
container.wire(modules=["myapp.views"])
随后到 INSTALLED_APPS 中的应用配置里添加您的应用。
# settings.py
INSTALLED_APPS = [
'myapp.apps.MyAppConfig',
...
]
步骤 6:测试视图
启动 Django 服务器,并通过 URL http://localhost:8000/greet/?name=Django 访问视图。
您应当会看到如下内容。

FastAPI 中的依赖注入
FastAPI 通过特殊标记 Depends() 提供内置的依赖注入支持,可注入函数返回的任何内容。
以下是 FastAPI 依赖注入的分步指南。
步骤 1:定义服务
class GreetingService:
def get_greeting(self, name: str) -> str:
return f"Hello, {name}!"
步骤 2:创建依赖函数
def get_greeting_service():
return GreetingService()
步骤 3:设置 FastAPI 应用并使用依赖注入
app = FastAPI()
@app.get("/greet")
def greet(
name: str = "World", service: GreetingService = Depends(get_greeting_service)
):
return {"message": service.get_greeting(name)}
步骤 4:运行并测试应用
使用命令 uvicorn main:app –reload 启动服务器,并访问 URL http://localhost:8000/greet?name=FastAPI。您应当会看到如下响应。

Python 依赖注入框架对比
有多种 Python 依赖注入框架,各有优劣、语法和适用场景。选择合适的框架取决于项目的规模与结构。
以下是几个常见 Python 依赖注入框架的对比。
|
框架 |
最佳适用场景 |
优势 |
局限 |
|
|
生产环境的复杂应用 |
功能全面、性能快、可配置性强 |
相对繁琐 |
|
简单、适合最小化应用 |
语法简洁 |
功能不如其他框架丰富 |
|
|
高级绑定逻辑 |
自动装配 |
由于自省而较慢 |
|
|
简单应用 |
精简且快速 |
与其他框架相比功能有限 |
|
|
FastAPI 应用 |
内置支持异步,也便于测试 |
无法在 FastAPI 之外复用 |
|
|
Flask 应用 |
易于与 Flask 集成 |
依赖 |
|
|
Django 项目 |
为视图与中间件提供依赖注入 |
依赖 |
高级实现模式
随着应用代码库变大,依赖注入的维护将变得具有挑战性。
您可以使用高级依赖注入来处理请求作用域资源、清理和关注点分离等任务。
作用域依赖
作用域依赖让您能够控制依赖的存续时间,例如:为每个请求创建新对象、在应用生命周期内共享对象,或对数据库连接等资源进行自动清理。
大多数依赖注入框架支持以下作用域:
- 单例(Singleton): 整个应用只有一个实例。
- 工厂(Factory): 每次注入时创建一个新实例。
- 线程本地或请求级: 每个线程/请求一个实例。
- 资源(Resource): 带有初始化与清理逻辑的受管资源。
下面是使用 dependency-injector 进行嵌套容器与作用域依赖的示例。
from dependency_injector import containers, providers, resources
# A simple resource that needs setup and cleanup
class MyResource:
def __enter__(self):
print("Setting up the resource")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("Cleaning up the resource")
# Create a container to manage the resource
class MyContainer(containers.DeclarativeContainer):
resource = providers.Resource(MyResource)
# Using the resource with a context manager
container = MyContainer()
with container.resource() as r:
print("Using the resource")
在上述示例中,MyResource 模拟了一个需要初始化与清理的类,例如打开/关闭文件或数据库。
container 通过 providers.Resource 提供资源,在使用 with container.resource() 时,会正确进入与退出上下文。
该模式确保文件句柄、数据库连接或网络会话等资源得到干净的管理。
设想一个需要为当前请求提供数据库会话的 API 端点,或带有请求上下文的记录器。
借助作用域依赖,框架会在请求开始时创建请求作用域的容器,并从该作用域解析所有依赖。
当请求结束时,会自动进行必要的清理,例如关闭数据库会话。这确保了不同请求之间不会共享状态,并避免资源泄漏。
异步注入
为高效处理 I/O 密集型任务,需要异步执行。FastAPI 与 dependency-injector 等框架中的依赖注入系统支持使用 async/await 的异步注入。
异步注入指的是注入 async def 函数或管理 async resources(如异步数据库连接、外部 API 或后台任务)等依赖。
依赖注入系统不会同步解析它们,而是在注入前等待其解析完成。
以下是 FastAPI 中异步依赖的示例:
from fastapi import FastAPI, Depends
app = FastAPI()
# Async dependency
async def get_user():
# Simulate async I/O
await some_async_db_query()
return {"name": "Philip Jones"}
@app.get("/profile")
async def read_profile(user: dict = Depends(get_user)):
return {"user": user}
FastAPI 会自动处理 get_user 的异步解析并注入结果。
dependency-injector 也通过 AsyncResource 提供者支持异步生命周期管理。
from dependency_injector import containers, providers, resources
class AsyncDB:
async def connect(self):
print("Connected")
return self
async def disconnect(self):
print("Disconnected")
class AsyncDBResource(resources.AsyncResource):
async def init(self) -> AsyncDB:
db = AsyncDB()
return await db.connect()
async def shutdown(self, db: AsyncDB):
await db.disconnect()
class Container(containers.DeclarativeContainer):
db = providers.Resource(AsyncDBResource)
# Usage
async def main():
container = Container()
await container.init_resources()
db = await container.db() # Get the AsyncDB instance
# Use db here...
await container.shutdown_resources()
在上述代码中:
-
resources.AsyncResource用于异步的初始化/清理逻辑。 -
await container.init_resources()初始化所有资源并调用init。 -
await container.shutdown_resources()清理所有资源。 -
await container.db()获取实际的资源实例,即AsyncDB。
使用异步依赖注入可以避免在异步路由中进行同步 I/O,从而保持应用的高性能,并免去在业务逻辑中手动等待依赖的繁琐。
安全影响
尽管依赖注入提升了模块化与可测试性,但若管理不当,也可能引入安全风险。
一个重要风险是依赖混淆:恶意包伪装成合法包。例如,攻击者会将包上传至PyPI,使用企业内部系统中的同名包,诱导应用使用恶意版本。
为确保安全,请勿注入不受信任或用户控制的组件,并在注入前验证配置来源(如环境变量)。
使用 requirements.txt 或 poetry.lock 锁定精确版本,并在依赖变更时自动进行安全扫描。此外,定期审计依赖的漏洞。
集中管理依赖也很重要,因为这能确保应用间的一致性,并更容易应用安全补丁或版本更新。还能避免在代码库中以不同方式注入的“影子依赖”。
多种工具可用于发现风险并改善安全状况:
- Snyk: 扫描 Python 包中的已知漏洞并给出修复建议。
- Xygeni: 为 CI/CD 流水线提供防护。
- Safety: 检查不安全包与代码漏洞。
- Bandit: 面向安全问题的 Python 静态代码分析器。
真实世界的应用
依赖注入并非只是理论模式;在各类生产系统中,它被用来管理复杂性、提升灵活性并简化测试。
Web 服务配置
在现代 Web 应用中,依赖注入在管理认证、日志记录和数据库访问等常见服务方面至关重要。
像 FastAPI 这样的框架在路由和依赖解析中发挥关键作用。
下面的示例展示了在 FastAPI 中集中化的认证实现。
from fastapi import Depends, HTTPException
def get_current_user(token: str = Depends(oauth2_scheme)):
user = verify_token(token)
if not user:
raise HTTPException(status_code=401)
return user
@app.get("/protected")
def protected_route(user: dict = Depends(get_current_user)):
return {"message": f"Hello {user['name']}"}
在上述代码中,get_current_user() 被作为路由依赖注入,但校验令牌的逻辑是集中且可复用的。
数据库访问也可以采用同样方式,使数据库会话得到自动管理与清理。
以下示例展示如何使用依赖注入集中管理数据库访问。
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db_session() -> AsyncSession:
async with async_session() as session:
yield session
@app.get("/items")
async def read_items(db: AsyncSession = Depends(get_db_session)):
return await db.execute("SELECT * FROM items")
在上述代码中,get_db_session 是依赖提供者。当调用 read_items 路由时,FastAPI 会自动注入 db 参数,使路由无需手动管理会话的初始化或清理即可访问数据库。
测试驱动开发
依赖注入让您可以轻松更改组件接收的依赖,这对测试而言非常重要。
在测试中,您可以用模拟对象替换真实服务,使代码更加模块化、易于测试。
FastAPI 与 dependency-injector 支持在测试时覆盖依赖。
下面是 FastAPI 的示例。
from fastapi.testclient import TestClient
from main import app, get_db_session
def override_db():
return TestDBSession()
app.dependency_overrides[get_db_session] = override_db
client = TestClient(app)
response = client.get("/items")
assert response.status_code == 200
在上述代码示例中,get_db_session 在测试期间被模拟数据库覆盖;因此无需修补或更改生产代码。
同样在 depedency-injector 中,您也可以在测试时用模拟数据库覆盖真实数据库,如下例所示。
from containers import Container
def test_service_behavior():
container = Container()
container.db.override(providers.Factory(TestDB))
service = container.service()
assert service.get_data() == "mocked"
override() 方法会用模拟替换真实依赖,干净、可控且可回滚。
以下是进行依赖覆盖时应采用的一些策略:
- 构造函数注入: 将模拟对象直接传给构造函数。
- 框架级覆盖: 使用内置覆盖工具,如 FastAPI 的
FastAPI.dependency_overrides或dependency-injector中的dependency_injector.override()。 - Fixtures: 使用
pytest等测试框架创建可复用的模拟对象,并通过 fixtures 注入。
最佳实践与反模式
虽然依赖注入旨在让代码模块化、易测试,但也可能引发缺陷或增加复杂性。
推荐实践
创建依赖时应遵循以下最佳实践:
- 面向接口编程,而非实现: 使用抽象基类或接口定义依赖,而非复杂实现。这样更易测试,也鼓励在单元测试中使用模拟。
- 多用组合,少用继承: 相比构建复杂的类层次结构,使用注入的依赖组合对象,使组件小而专注。
- 集中式依赖配置: 使用单一容器或模块定义并管理依赖,便于覆盖与测试。
- 避免循环依赖: 通过使用工厂函数或提供者延迟实例化,谨慎设计依赖图,防止循环引用。
常见陷阱
使用依赖注入时应避免以下常见陷阱。
- 服务定位器反模式: 避免使用全局容器动态获取依赖,这会掩盖依赖关系,使代码难以理解和测试。
- 过度注入: 避免向单个类或函数注入过多服务,以免构造函数臃肿、逻辑难测。
- 作用域管理不当与状态泄漏: 始终使用正确的作用域,避免在用户或测试之间共享状态。
Python 中的依赖注入测试
依赖注入便于将模拟对象注入组件,从而简化单元测试。这使您可以将核心组件与测试专用组件分离。
借助依赖注入,测试具有确定性、无副作用,且执行更快。
下面示例展示如何在测试中注入这些依赖的 mock、stub 或 fake 版本。
class EmailService:
def send_email(self, to: str, message: str): ...
class Notifier:
def __init__(self, email_service: EmailService):
self.email_service = email_service
def alert(self, user_email):
self.email_service.send_email(user_email, "Alert!")
在测试中,用 mock 替换 EmailService:
class TestEmailService:
def send_email(self, to, message):
self.sent_to = to
def test_notifier_sends_email():
test_email = TestEmailService()
notifier = Notifier(test_email)
notifier.alert("test@example.com")
assert test_email.sent_to == "test@example.com"
请务必经常使用依赖注入,避免使用猴子补丁(monkey patching),以提升可维护性并避免内部名称变更导致的破坏性影响。
结语
对于小项目,您一开始或许不需要依赖注入,但随着项目增长与扩展,依赖注入带来的结构化与模块化优势将愈发明显。
展望未来,Python 中的依赖注入前景可期,我们有望看到更完善的类型系统与增强的异步生命周期管理。
掌握依赖注入是成为熟练 Python 开发者的关键一步。但学习并不止于此;还有许多高级概念可以进一步提升您的 Python 技能。
为加速您的学习之旅,以下是一些值得一读的资源。
课程:
- Developing Python Packages
- Introduction to PySpark
- Object-Oriented Programming in Python
- Intermediate Object-Oriented Programming in Python
博文:
FAQs
什么是依赖注入?
依赖注入是一种设计模式,它让您通过从外部传入类所需的依赖来编写更整洁的代码,而不是在类内部创建依赖,从而让代码更易管理与修改。
依赖注入的主要好处是什么?
依赖注入使代码更具模块化、可测试性和可维护性。
如何在 Python 中实现依赖注入?
您可以在 Python 中通过手动方式,或使用诸如 dependency_injector 等库来实现依赖注入。
什么是异步注入?
异步注入指的是注入 async def 函数或管理 async resources(如异步数据库连接、外部 API 或后台任务)等依赖。
依赖注入有哪些安全风险?如何应对?
一个显著风险是依赖混淆,即恶意包伪装成合法包。为防止此类问题,您可以使用 Snyk、Xygeni、Safety 或 Bandit 等工具扫描代码与包的漏洞。
