Python 依赖注入原理与实现：解耦你的代码
引言
大家好，我是一名正在从Rust转向Python的后端开发者。在大型项目开发中，代码的可维护性和可测试性是非常重要的。依赖注入（Dependency Injection）是一种设计模式，可以帮助我们实现代码解耦，提高代码的可测试性和可维护性。今天，我想和大家分享一下我在Python中实现依赖注入的经验。
什么是依赖注入？
概念
依赖注入是一种软件设计模式，它允许我们将对象的依赖关系从对象内部转移到外部。这样做的好处是：
- 解耦：降低模块之间的耦合度
- 可测试性：便于单元测试，可以注入mock对象
- 可维护性：更容易替换和扩展组件
- 可配置性：可以通过配置来改变依赖关系
传统方式 vs 依赖注入
# 传统方式：硬编码依赖 class Database: def connect(self): print("连接数据库") class UserService: def __init__(self): # 硬编码依赖 self.db = Database() def get_user(self, user_id): self.db.connect() return f"用户 {user_id}" # 依赖注入：依赖由外部提供 class UserServiceDI: def __init__(self, db): # 依赖通过构造函数注入 self.db = db def get_user(self, user_id): self.db.connect() return f"用户 {user_id}" # 使用 db = Database() service = UserServiceDI(db)依赖注入的实现方式
1. 构造函数注入
class Logger: def log(self, message): print(f"日志: {message}") class EmailService: def send_email(self, to, subject, body): print(f"发送邮件到 {to}: {subject}") class UserService: def __init__(self, logger: Logger, email_service: EmailService): self.logger = logger self.email_service = email_service def register_user(self, email): self.logger.log(f"注册用户: {email}") self.email_service.send_email(email, "欢迎注册", "感谢您的注册") # 使用 logger = Logger() email_service = EmailService() user_service = UserService(logger, email_service) user_service.register_user("test@example.com")2. 属性注入
class UserService: def __init__(self): self.logger = None self.email_service = None def set_logger(self, logger): self.logger = logger def set_email_service(self, email_service): self.email_service = email_service def register_user(self, email): if self.logger: self.logger.log(f"注册用户: {email}") if self.email_service: self.email_service.send_email(email, "欢迎注册", "感谢您的注册") # 使用 user_service = UserService() user_service.set_logger(Logger()) user_service.set_email_service(EmailService()) user_service.register_user("test@example.com")3. 方法注入
class UserService: def register_user(self, email, logger: Logger, email_service: EmailService): logger.log(f"注册用户: {email}") email_service.send_email(email, "欢迎注册", "感谢您的注册") # 使用 logger = Logger() email_service = EmailService() user_service = UserService() user_service.register_user("test@example.com", logger, email_service)依赖注入容器
简单的依赖注入容器
class Container: def __init__(self): self._services = {} def register(self, name, service): self._services[name] = service def resolve(self, name): return self._services.get(name) # 使用 container = Container() container.register('logger', Logger()) container.register('email_service', EmailService()) logger = container.resolve('logger') email_service = container.resolve('email_service') user_service = UserService(logger, email_service)高级依赖注入容器
from typing import Type, Dict, Any class DIContainer: def __init__(self): self._factories: Dict[Type, callable] = {} self._instances: Dict[Type, Any] = {} def register(self, interface: Type, factory): self._factories[interface] = factory def register_singleton(self, interface: Type, instance): self._instances[interface] = instance def resolve(self, interface: Type) -> Any: # 检查是否有单例实例 if interface in self._instances: return self._instances[interface] # 检查是否有工厂函数 if interface in self._factories: return self._factories[interface]() raise ValueError(f"未注册的依赖: {interface}") # 使用 container = DIContainer() container.register_singleton(Logger, Logger()) container.register_singleton(EmailService, EmailService()) container.register(UserService, lambda: UserService( container.resolve(Logger), container.resolve(EmailService) )) user_service = container.resolve(UserService) user_service.register_user("test@example.com")使用第三方依赖注入库
injector库
from injector import Injector, inject, singleton class Logger: def log(self, message): print(f"日志: {message}") class EmailService: def send_email(self, to, subject, body): print(f"发送邮件到 {to}: {subject}") class UserService: @inject def __init__(self, logger: Logger, email_service: EmailService): self.logger = logger self.email_service = email_service def register_user(self, email): self.logger.log(f"注册用户: {email}") self.email_service.send_email(email, "欢迎注册", "感谢您的注册") # 创建注入器 injector = Injector() # 获取服务 user_service = injector.get(UserService) user_service.register_user("test@example.com")autowire库
from autowire import autowire, wire class Logger: def log(self, message): print(f"日志: {message}") class EmailService: def send_email(self, to, subject, body): print(f"发送邮件到 {to}: {subject}") @autowire class UserService: def __init__(self, logger: Logger, email_service: EmailService): self.logger = logger self.email_service = email_service def register_user(self, email): self.logger.log(f"注册用户: {email}") self.email_service.send_email(email, "欢迎注册", "感谢您的注册") # 使用 with wire(): user_service = UserService() user_service.register_user("test@example.com")实战项目：完整的依赖注入系统
from typing import Type, Dict, Any, Optional from abc import ABC, abstractmethod # 定义接口 class ILogger(ABC): @abstractmethod def log(self, message: str): pass class IEmailService(ABC): @abstractmethod def send_email(self, to: str, subject: str, body: str): pass class IUserRepository(ABC): @abstractmethod def save(self, user: dict): pass # 实现类 class ConsoleLogger(ILogger): def log(self, message: str): print(f"[控制台日志] {message}") class FileLogger(ILogger): def __init__(self, file_path: str): self.file_path = file_path def log(self, message: str): with open(self.file_path, 'a') as f: f.write(f"{message}\n") class SmtpEmailService(IEmailService): def __init__(self, smtp_server: str, smtp_port: int): self.smtp_server = smtp_server self.smtp_port = smtp_port def send_email(self, to: str, subject: str, body: str): print(f"通过SMTP发送邮件到 {to}: {subject}") class MockEmailService(IEmailService): def send_email(self, to: str, subject: str, body: str): print(f"[Mock] 发送邮件到 {to}: {subject}") class DatabaseUserRepository(IUserRepository): def __init__(self, db_url: str): self.db_url = db_url def save(self, user: dict): print(f"保存用户到数据库: {user}") # 依赖注入容器 class DIContainer: def __init__(self, environment: str = "development"): self.environment = environment self._factories: Dict[Type, callable] = {} self._instances: Dict[Type, Any] = {} self._configure() def _configure(self): if self.environment == "development": self.register_singleton(ILogger, ConsoleLogger()) self.register_singleton(IEmailService, MockEmailService()) self.register_singleton(IUserRepository, DatabaseUserRepository("sqlite:///dev.db")) else: self.register_singleton(ILogger, FileLogger("app.log")) self.register_singleton(IEmailService, SmtpEmailService("smtp.example.com", 587)) self.register_singleton(IUserRepository, DatabaseUserRepository("postgresql://prod")) def register_singleton(self, interface: Type, instance: Any): self._instances[interface] = instance def resolve(self, interface: Type) -> Any: if interface in self._instances: return self._instances[interface] raise ValueError(f"未注册的依赖: {interface}") # 业务服务 class UserService: def __init__(self, logger: ILogger, email_service: IEmailService, user_repo: IUserRepository): self.logger = logger self.email_service = email_service self.user_repo = user_repo def register_user(self, email: str, password: str): self.logger.log(f"开始注册用户: {email}") user = { "email": email, "password": password, "created_at": "2026-05-08" } self.user_repo.save(user) self.email_service.send_email(email, "欢迎注册", "感谢您的注册") self.logger.log(f"用户注册成功: {email}") # 使用 container = DIContainer(environment="development") logger = container.resolve(ILogger) email_service = container.resolve(IEmailService) user_repo = container.resolve(IUserRepository) user_service = UserService(logger, email_service, user_repo) user_service.register_user("test@example.com", "password123")依赖注入的最佳实践
1. 使用接口抽象
from abc import ABC, abstractmethod class ICache(ABC): @abstractmethod def get(self, key: str) -> Optional[str]: pass @abstractmethod def set(self, key: str, value: str, ttl: Optional[int] = None): pass class RedisCache(ICache): def get(self, key: str) -> Optional[str]: # Redis实现 pass def set(self, key: str, value: str, ttl: Optional[int] = None): # Redis实现 pass class MemoryCache(ICache): def __init__(self): self._cache = {} def get(self, key: str) -> Optional[str]: return self._cache.get(key) def set(self, key: str, value: str, ttl: Optional[int] = None): self._cache[key] = value2. 使用类型提示
from typing import Type class Container: def __init__(self): self._services: Dict[Type, Any] = {} def register(self, interface: Type, implementation: Any): self._services[interface] = implementation def resolve(self, interface: Type) -> Any: return self._services.get(interface)3. 分层注入
class Config: def __init__(self, db_url: str, api_key: str): self.db_url = db_url self.api_key = api_key class Database: def __init__(self, config: Config): self.url = config.db_url class ApiClient: def __init__(self, config: Config): self.api_key = config.api_key class Service: def __init__(self, db: Database, api: ApiClient): self.db = db self.api = api # 注册顺序很重要 container = Container() container.register(Config, Config("sqlite:///db", "secret")) container.register(Database, Database(container.resolve(Config))) container.register(ApiClient, ApiClient(container.resolve(Config))) container.register(Service, Service( container.resolve(Database), container.resolve(ApiClient) ))与Rust依赖注入的对比
| 特性 | Python | Rust |
|---|---|---|
| 类型安全 | 通过类型提示 | 原生类型系统 |
| 编译时检查 | 运行时 | 编译时 |
| 第三方库 | injector, autowire | di, bevy_ecs |
| 宏支持 | 有限 | 强大的宏系统 |
| 性能 | 运行时开销 | 零运行时开销 |
总结
依赖注入是一种强大的设计模式，可以帮助我们：
- 降低耦合度：模块之间通过接口交互
- 提高可测试性：可以轻松替换依赖为mock对象
- 提高可维护性：更容易扩展和修改代码
- 提高可配置性：可以通过配置改变依赖关系
在Python中实现依赖注入可以使用：
- 手动实现简单的依赖注入容器
- 使用第三方库如injector、autowire
- 利用类型提示提高代码质量
作为从Rust转向Python的开发者，我发现Python的依赖注入虽然没有Rust的编译时保证，但通过合理使用类型提示和第三方库，可以达到很好的效果。
延伸阅读：
- injector官方文档
- autowire官方文档
- 依赖注入原理
