Python 依赖注入原理与实现：解耦你的代码-171主机测评

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] = value

2. 使用类型提示

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, autowire	di, bevy_ecs
宏支持	有限	强大的宏系统
性能	运行时开销	零运行时开销

总结

依赖注入是一种强大的设计模式，可以帮助我们：

降低耦合度：模块之间通过接口交互

提高可测试性：可以轻松替换依赖为mock对象

提高可维护性：更容易扩展和修改代码

提高可配置性：可以通过配置改变依赖关系

在Python中实现依赖注入可以使用：

手动实现简单的依赖注入容器
使用第三方库如injector、autowire
利用类型提示提高代码质量

作为从Rust转向Python的开发者，我发现Python的依赖注入虽然没有Rust的编译时保证，但通过合理使用类型提示和第三方库，可以达到很好的效果。

Python 依赖注入原理与实现：解耦你的代码