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(testexample.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(testexample.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(testexample.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(testexample.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(testexample.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(testexample.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(environmentdevelopment) 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(testexample.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依赖注入的对比特性PythonRust类型安全通过类型提示原生类型系统编译时检查运行时编译时第三方库injector, autowiredi, bevy_ecs宏支持有限强大的宏系统性能运行时开销零运行时开销总结依赖注入是一种强大的设计模式可以帮助我们降低耦合度模块之间通过接口交互提高可测试性可以轻松替换依赖为mock对象提高可维护性更容易扩展和修改代码提高可配置性可以通过配置改变依赖关系在Python中实现依赖注入可以使用手动实现简单的依赖注入容器使用第三方库如injector、autowire利用类型提示提高代码质量作为从Rust转向Python的开发者我发现Python的依赖注入虽然没有Rust的编译时保证但通过合理使用类型提示和第三方库可以达到很好的效果。延伸阅读injector官方文档autowire官方文档依赖注入原理