跳至内容
Python

Python 依赖注入:构建模块化、可测试的代码

了解并实现多种 Python 依赖注入框架,让您的代码更模块化、可测试且易维护。
Adejumo Ridwan Suleiman's photo
Adejumo Ridwan Suleiman
已更新 · 14分钟 阅读

使用 AI 探索

ChatGPTClaudePerplexity

创建您的免费账户

实践并学习路径你的进步。

使用 Google 继续
继续操作即表示您接受我们的《使用条款》和《隐私政策》,并同意您的数据存储在美国。

依赖注入是一种设计模式,它将对象或服务(即依赖)从外部提供给类,而不是由类在内部自行创建。这样可以解耦组件。

在 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。

您应当会看到如下内容。

Image showing the url of a running Flask appliction

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 访问视图。

您应当会看到如下内容。

Image showing the Django application running.

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。您应当会看到如下响应。

Image showing a running FastAPI application.

Python 依赖注入框架对比

有多种 Python 依赖注入框架,各有优劣、语法和适用场景。选择合适的框架取决于项目的规模与结构。

以下是几个常见 Python 依赖注入框架的对比。

框架

最佳适用场景

优势

局限

dependency-injector

生产环境的复杂应用

功能全面、性能快、可配置性强

相对繁琐

injector

简单、适合最小化应用

语法简洁

功能不如其他框架丰富

pinject

高级绑定逻辑

自动装配

由于自省而较慢

punq

简单应用

精简且快速

与其他框架相比功能有限

FastAPI Depends()

FastAPI 应用

内置支持异步,也便于测试

无法在 FastAPI 之外复用

Flask-injector

Flask 应用

易于与 Flask 集成

依赖 injector

django-injector

Django 项目

为视图与中间件提供依赖注入

依赖 injector

高级实现模式

随着应用代码库变大,依赖注入的维护将变得具有挑战性。

您可以使用高级依赖注入来处理请求作用域资源、清理和关注点分离等任务。

作用域依赖

作用域依赖让您能够控制依赖的存续时间,例如:为每个请求创建新对象、在应用生命周期内共享对象,或对数据库连接等资源进行自动清理。

大多数依赖注入框架支持以下作用域:

  • 单例(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 技能。

为加速您的学习之旅,以下是一些值得一读的资源。

课程:

博文:


Adejumo Ridwan Suleiman's photo
Author
Adejumo Ridwan Suleiman
LinkedIn

资深数据科学讲师兼生物统计学家,精通 Python、R 和机器学习。

FAQs

什么是依赖注入?

依赖注入是一种设计模式,它让您通过从外部传入类所需的依赖来编写更整洁的代码,而不是在类内部创建依赖,从而让代码更易管理与修改。

依赖注入的主要好处是什么?

依赖注入使代码更具模块化、可测试性和可维护性。

如何在 Python 中实现依赖注入?

您可以在 Python 中通过手动方式,或使用诸如 dependency_injector 等库来实现依赖注入。

什么是异步注入?

异步注入指的是注入 async def 函数或管理 async resources(如异步数据库连接、外部 API 或后台任务)等依赖。

依赖注入有哪些安全风险?如何应对?

一个显著风险是依赖混淆,即恶意包伪装成合法包。为防止此类问题,您可以使用 Snyk、Xygeni、Safety 或 Bandit 等工具扫描代码与包的漏洞。

主题
Python

与 DataCamp 一起学习 Python

课程

Python for Developers 入门

3 小时
183.9K
掌握 Python 编程基础。 无需任何先验知识!
查看详情开始课程

创建您的免费账户

实践并学习路径你的进步。

使用 Google 继续
继续操作即表示您接受我们的《使用条款》和《隐私政策》,并同意您的数据存储在美国。
查看更多