什么是依赖注入，为什么在 Python 中应该使用它？

依赖注入是一种设计模式，它通过从外部来源而非内部创建的方式向组件提供依赖项。在 Python 中，它通过允许注入模拟依赖项来提高可测试性，通过使依赖项显式化来增强可维护性，并促进组件之间的松耦合。在构建 API、服务或具有复杂依赖关系图的应用程序时，它尤其有价值。

我应该使用像 dependency-injector 这样的 DI 框架，还是手动实现 DI？

对于小型到中型应用，手动构造函数注入通常已经足够，并且可以保持代码的简洁。当存在复杂的依赖关系图、众多依赖项或需要自动解析依赖项时，建议使用如 dependency-injector 或 injector 这样的框架。对于大多数 Python 项目，推荐使用手动依赖注入，因为它明确且能够利用 Python 的动态特性，而不会增加框架的开销。

如何在 Python 中实现构造函数注入？

构造函数注入涉及在 init 方法中将依赖项作为参数接受。例如， def __init__(self, repo: UserRepository, logger: Logger) 创建了一个具有注入依赖项的服务。这使得依赖关系显式化，并确保没有满足所需依赖项的情况下无法实例化组件。Python 的类型提示使这种模式更加清晰。

依赖注入和依赖倒置有什么区别？

依赖倒置原则（DIP）是一种设计原则，表明高层模块不应依赖于低层模块，两者都应依赖于抽象（协议或ABC）。依赖注入是一种实现技术，通过从外部提供依赖关系，使DIP成为可能。在Python中，你可以通过定义协议或抽象基类，并使用DI注入具体实现来实现DIP。

在什么情况下我应该避免在 Python 中使用依赖注入？

在直接实例化更清晰的情况下，例如创建没有外部依赖的基本数据类，或当使用依赖注入的开销超过其好处时，应避免使用依赖注入（DI）。不要注入那些真正属于内部实现细节的依赖项。此外，在小型实用工具或一次性脚本中，应避免过度设计，以免依赖注入带来不必要的复杂性。

依赖注入如何改进 Python 中的测试？

依赖注入（DI）通过允许你注入依赖项的模拟或测试实现，使测试变得简单。你无需依赖真实的数据库、HTTP 客户端或外部服务，而是可以使用 unittest.mock 或 pytest 固件注入测试替身。这使单元测试更加快速、可靠，并且独立于外部系统。你可以在不搭建复杂测试基础设施的情况下，独立测试业务逻辑。

依赖注入：一种 Python 方式

Python 依赖注入模式实现清晰可测试的代码

依赖注入 (DI) 是一种基本的设计模式，它在 Python 应用程序中促进干净、可测试和易于维护的代码。

无论你是使用 FastAPI 构建 REST API，实现单元测试，还是使用 AWS Lambda 函数，理解依赖注入都将显著提高你的代码质量。

Python 包

什么是依赖注入？

依赖注入是一种设计模式，其中组件从外部源接收其依赖项，而不是内部创建它们。这种方法解耦了组件，使你的代码更加模块化、可测试和易于维护。

在 Python 中，依赖注入尤其强大，因为该语言的动态特性以及对协议、抽象基类和鸭子类型的广泛支持。Python 的灵活性意味着你可以在不使用重型框架的情况下实现 DI 模式，尽管在需要时也有可用的框架。

为什么在 Python 中使用依赖注入？

提高可测试性：通过注入依赖项，你可以轻松地用模拟对象或测试替身替换真实实现。这允许你编写快速、隔离的单元测试，不需要数据库或 API 等外部服务。在编写全面的单元测试时，依赖注入使得用测试替身替换真实依赖变得轻而易举。

更好的可维护性：依赖项在你的代码中变得显式。当你查看一个构造函数时，你立即知道一个组件需要什么。这使得代码库更容易理解和修改。

松耦合：组件依赖于抽象（协议或 ABC），而不是具体的实现。这意味着你可以更改实现而不会影响依赖代码。

灵活性：你可以为不同环境（开发、测试、生产）配置不同的实现，而无需更改业务逻辑。这在将 Python 应用程序部署到不同平台时尤其有用，无论是 AWS Lambda 还是传统服务器。

构造函数注入：Python 的方式

在 Python 中实现依赖注入最常见的也是最符合 Python 习惯的方式是通过构造函数注入——在 __init__ 方法中将依赖项作为参数接受。

基本示例

以下是一个演示构造函数注入的简单示例：

from typing import Protocol
from abc import ABC, abstractmethod

# 定义一个仓库的协议
class UserRepository(Protocol):
    def find_by_id(self, user_id: int) -> 'User | None':
        ...
    
    def save(self, user: 'User') -> 'User':
        ...

# 服务依赖于仓库协议
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo
    
    def get_user(self, user_id: int) -> 'User | None':
        return self.repo.find_by_id(user_id)

这种模式清楚地表明 UserService 需要一个 UserRepository。没有提供仓库，你就不能创建 UserService，这防止了因缺失依赖而导致的运行时错误。

多个依赖项

当组件有多个依赖项时，只需将它们作为构造函数参数添加：

class EmailService(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

class Logger(Protocol):
    def info(self, msg: str) -> None:
        ...
    
    def error(self, msg: str, err: Exception) -> None:
        ...

class OrderService:
    def __init__(
        self,
        repo: OrderRepository,
        email_svc: EmailService,
        logger: Logger,
        payment_svc: PaymentService,
    ):
        self.repo = repo
        self.email_svc = email_svc
        self.logger = logger
        self.payment_svc = payment_svc

使用协议和抽象基类

实现依赖注入时的一个关键原则是 依赖倒置原则（DIP）：高层模块不应依赖于低层模块；两者都应依赖于抽象。

在 Python 中，你可以使用协议（结构化类型）或 抽象基类（ABCs）（命名类型）来定义抽象。

协议（Python 3.8+）

协议使用结构化类型——如果对象具有所需的方法，它就满足协议：

from typing import Protocol

class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

# 任何具有 process_payment 方法的类都满足此协议
class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # 信用卡逻辑
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # PayPal 逻辑
        return True

# 服务接受任何 PaymentProcessor
class OrderService:
    def __init__(self, payment_processor: PaymentProcessor):
        self.payment_processor = payment_processor

抽象基类

ABCs 使用命名类型——类必须显式继承自 ABC：

from abc import ABC, abstractmethod

class PaymentProcessor(ABC):
    @abstractmethod
    def process_payment(self, amount: float) -> bool:
        pass

class CreditCardProcessor(PaymentProcessor):
    def process_payment(self, amount: float) -> bool:
        return True

何时使用协议 vs ABCs：当想要结构化类型和灵活性时使用协议。当需要强制继承层次结构或提供默认实现时使用 ABCs。

实际示例：数据库抽象

在 Python 应用程序中处理数据库时，你通常需要抽象数据库操作。以下是依赖注入如何帮助的示例：

from typing import Protocol, Optional
from contextlib import contextmanager

class Database(Protocol):
    @contextmanager
    def transaction(self):
        ...
    
    def execute(self, query: str, params: dict) -> None:
        ...
    
    def fetch_one(self, query: str, params: dict) -> Optional[dict]:
        ...

# 仓库依赖于抽象
class UserRepository:
    def __init__(self, db: Database):
        self.db = db
    
    def find_by_id(self, user_id: int) -> Optional['User']:
        result = self.db.fetch_one(
            "SELECT * FROM users WHERE id = :id",
            {"id": user_id}
        )
        if result:
            return User(**result)
        return None

这种模式允许你在不更改仓库代码的情况下交换数据库实现（PostgreSQL、SQLite、MongoDB）。

组合根模式

组合根是在应用程序的入口点（通常是 main.py 或你的应用程序工厂）组装所有依赖项的地方。这集中了依赖项配置并使依赖图显式化。

def create_app() -> FastAPI:
    app = FastAPI()
    
    # 初始化基础设施依赖项
    db = init_database()
    logger = init_logger()
    
    # 初始化仓库
    user_repo = UserRepository(db)
    order_repo = OrderRepository(db)
    
    # 使用依赖项初始化服务
    email_svc = EmailService(logger)
    payment_svc = PaymentService(logger)
    user_svc = UserService(user_repo, logger)
    order_svc = OrderService(order_repo, email_svc, logger, payment_svc)
    
    # 初始化 HTTP 处理器
    user_handler = UserHandler(user_svc)
    order_handler = OrderHandler(order_svc)
    
    # 连接路由
    app.include_router(user_handler.router)
    app.include_router(order_handler.router)
    
    return app

这种方法使你的应用程序结构和依赖项来源清晰可见。在构建遵循整洁架构原则的应用程序时，这种模式特别有价值，因为需要协调多个依赖层。

依赖注入框架

对于具有复杂依赖图的大型应用程序，手动管理依赖项可能会变得繁琐。Python 有几种 DI 框架可以帮助你：

Dependency Injector

Dependency Injector 是一个流行的框架，提供基于容器的依赖注入方法。

安装：

pip install dependency-injector

示例：

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

class Container(containers.DeclarativeContainer):
    # 配置
    config = providers.Configuration()
    
    # 数据库
    db = providers.Singleton(
        Database,
        connection_string=config.database.url
    )
    
    # 仓库
    user_repository = providers.Factory(
        UserRepository,
        db=db
    )
    
    # 服务
    user_service = providers.Factory(
        UserService,
        repo=user_repository
    )

# 使用
container = Container()
container.config.database.url.from_env("DATABASE_URL")
user_service = container.user_service()

Injector

Injector 是一个受 Google Guice 启发的轻量级库，专注于简洁性。

安装：

pip install injector

示例：

from injector import Injector, inject, Module, provider

class DatabaseModule(Module):
    @provider
    def provide_db(self) -> Database:
        return Database(connection_string=". . .")

class UserModule(Module):
    @inject
    def __init__(self, repo: UserRepository):
        self.repo = repo

injector = Injector([DatabaseModule()])
user_service = injector.get(UserService)

何时使用框架

使用框架时：

你的依赖图复杂，有许多相互依赖的组件
你有多个同一接口的实现，需要根据配置选择
你希望自动解析依赖项
你正在构建一个大型应用程序，手动连接变得容易出错

使用手动 DI 时：

你的应用程序规模较小到中等
依赖图简单且易于理解
你希望保持依赖项最小化和显式化
你更喜欢显式代码而不是框架的“魔法”

使用依赖注入进行测试

依赖注入的一个主要好处是提高了可测试性。以下是 DI 如何使测试更简单的示例：

单元测试示例

from unittest.mock import Mock
import pytest

# 测试用的模拟实现
class MockUserRepository:
    def __init__(self):
        self.users = {}
        self.error = None
    
    def find_by_id(self, user_id: int) -> Optional['User']:
        if self.error:
            raise self.error
        return self.users.get(user_id)
    
    def save(self, user: 'User') -> 'User':
        if self.error:
            raise self.error
        self.users[user.id] = user
        return user

# 使用模拟进行测试
def test_user_service_get_user():
    mock_repo = MockUserRepository()
    mock_repo.users[1] = User(id=1, name="John", email="john@example.com")
    
    service = UserService(mock_repo)
    
    user = service.get_user(1)
    assert user is not None
    assert user.name == "John"

这个测试运行速度快，不需要数据库，并且在隔离环境中测试你的业务逻辑。在使用 Python 单元测试时，依赖注入使得创建测试替身和验证交互变得容易。

使用 pytest 固件

pytest 固件与依赖注入配合得非常好：

@pytest.fixture
def mock_user_repository():
    return MockUserRepository()

@pytest.fixture
def user_service(mock_user_repository):
    return UserService(mock_user_repository)

def test_user_service_get_user(user_service, mock_user_repository):
    user = User(id=1, name="John", email="john@example.com")
    mock_user_repository.users[1] = user
    
    result = user_service.get_user(1)
    assert result.name == "John"

常见模式和最佳实践

1. 使用接口隔离

保持协议和接口小而专注，只关注客户端实际需要的内容：

# 好：客户端只需要读取用户
class UserReader(Protocol):
    def find_by_id(self, user_id: int) -> Optional['User']:
        ...
    
    def find_by_email(self, email: str) -> Optional['User']:
        ...

# 分离的接口用于写入
class UserWriter(Protocol):
    def save(self, user: 'User') -> 'User':
        ...
    
    def delete(self, user_id: int) -> bool:
        ...

2. 在构造函数中验证依赖项

构造函数应验证依赖项，并在初始化失败时抛出清晰的错误：

class UserService:
    def __init__(self, repo: UserRepository):
        if repo is None:
            raise ValueError("user repository cannot be None")
        self.repo = repo

3. 使用类型提示

类型提示使依赖项显式化，并有助于 IDE 支持和静态类型检查：

from typing import Protocol, Optional

class UserService:
    def __init__(
        self,
        repo: UserRepository,
        logger: Logger,
        email_service: EmailService,
    ) -> None:
        self.repo = repo
        self.logger = logger
        self.email_service = email_service

4. 避免过度注入

不要注入真正内部实现细节的依赖项。如果一个组件创建和管理自己的辅助对象，那就可以了：

# 好：内部辅助对象不需要注入
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo
        # 内部缓存 - 不需要注入
        self._cache: dict[int, User] = {}

5. 文档化依赖项

使用 docstrings 文档化为什么需要依赖项以及任何约束：

class UserService:
    """UserService 处理与用户相关的业务逻辑。
    
    Args:
        repo: 用于数据访问的 UserRepository。如果在并发上下文中使用，必须是线程安全的。
        logger: 用于错误跟踪和调试的 Logger。
    """
    def __init__(self, repo: UserRepository, logger: Logger):
        self.repo = repo
        self.logger = logger

FastAPI 中的依赖注入

FastAPI 通过其 Depends 机制内置了对依赖注入的支持：

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

def get_user_repository() -> UserRepository:
    db = get_database()
    return UserRepository(db)

def get_user_service(
    repo: Annotated[UserRepository, Depends(get_user_repository)]
) -> UserService:
    return UserService(repo)

@app.get("/users/{user_id}")
def get_user(
    user_id: int,
    service: Annotated[UserService, Depends(get_user_service)]
):
    user = service.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404)
    return user

FastAPI 的依赖注入系统会自动处理依赖图，使得构建干净、可维护的 API 变得容易。

何时不使用依赖注入

依赖注入是一个强大的工具，但并不总是必要的：

跳过 DI 的情况：

简单的值对象或数据类
内部辅助函数或工具
一次性脚本或小工具
当直接实例化更清晰和简单时

不使用 DI 的示例：

# 简单的数据类 - 不需要 DI
from dataclasses import dataclass

@dataclass
class Point:
    x: float
    y: float

# 简单的工具 - 不需要 DI
def format_currency(amount: float) -> str:
    return f"${amount:.2f}"

与 Python 生态系统的集成

依赖注入与 Python 的其他模式和工具无缝集成。当你构建使用 Python 包或单元测试框架的应用程序时，可以将这些服务注入到你的业务逻辑中：

class ReportService:
    def __init__(
        self,
        pdf_generator: PDFGenerator,
        repo: ReportRepository,
        logger: Logger,
    ):
        self.pdf_generator = pdf_generator
        self.repo = repo
        self.logger = logger
    
    def generate_report(self, report_id: int) -> bytes:
        report_data = self.repo.get_by_id(report_id)
        pdf = self.pdf_generator.generate(report_data)
        self.logger.info(f"Generated report {report_id}")
        return pdf

这允许你在测试时交换实现或使用模拟对象。

异步依赖注入

Python 的 async/await 语法与依赖注入很好地配合：

from typing import Protocol
import asyncio

class AsyncUserRepository(Protocol):
    async def find_by_id(self, user_id: int) -> Optional['User']:
        ...
    
    async def save(self, user: 'User') -> 'User':
        ...

class AsyncUserService:
    def __init__(self, repo: AsyncUserRepository):
        self.repo = repo
    
    async def get_user(self, user_id: int) -> Optional['User']:
        return await self.repo.find_by_id(user_id)
    
    async def get_users_batch(self, user_ids: list[int]) -> list['User']:
        tasks = [self.repo.find_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]

结论

依赖注入是编写可维护、可测试 Python 代码的基石。通过遵循本文中概述的模式——构造函数注入、基于协议的设计和组合根模式——你将创建更容易理解、测试和修改的应用程序。

对于小型到中型应用程序，从手动构造函数注入开始，随着依赖图的增长，可以考虑使用如 dependency-injector 或 injector 这样的框架。记住目标是清晰度和可测试性，而不是为了复杂性而复杂。

如需更多 Python 开发资源，请查看我们的 Python 快速参考以快速查阅 Python 语法和常见模式。