Python依赖注入与控制反转

Python依赖注入与控制反转一、什么是依赖注入依赖注入DI是一种设计模式将对象的依赖从内部创建改为外部传入实现控制反转IoC。# 没有依赖注入紧耦合class OrderService:def __init__(self):self.db MySQLDatabase() # 硬编码依赖self.mailer SMTPMailer() # 硬编码依赖def create_order(self, order_data):order self.db.insert(orders, order_data)self.mailer.send(order.user_email, 订单已创建)return order# 使用依赖注入松耦合class OrderService:def __init__(self, db, mailer):self.db db # 外部注入self.mailer mailer # 外部注入def create_order(self, order_data):order self.db.insert(orders, order_data)self.mailer.send(order.user_email, 订单已创建)return order# 可以轻松替换依赖测试时用Mockservice OrderService(dbFakeDatabase(), mailerFakeMailer())二、构造函数注入from typing import Protocolclass Database(Protocol):def query(self, sql: str) - list: ...def insert(self, table: str, data: dict) - dict: ...class EmailSender(Protocol):def send(self, to: str, subject: str, body: str) - None: ...class Logger(Protocol):def info(self, message: str) - None: ...def error(self, message: str) - None: ...class UserService:通过构造函数注入所有依赖def __init__(self, db: Database, email: EmailSender, logger: Logger):self.db dbself.email emailself.logger loggerdef register(self, name: str, email_addr: str) - dict:self.logger.info(f注册用户: {name})user self.db.insert(users, {name: name, email: email_addr})self.email.send(email_addr, 欢迎, f欢迎 {name}!)return userdef get_user(self, user_id: int) - dict:users self.db.query(fSELECT * FROM users WHERE id {user_id})if not users:raise ValueError(f用户 {user_id} 不存在)return users[0]三、简单的DI容器class Container:简单的依赖注入容器def __init__(self):self._factories {}self._singletons {}self._instances {}def register(self, interface, factory, singletonFalse):注册依赖self._factories[interface] factoryif singleton:self._singletons[interface] Truedef register_instance(self, interface, instance):注册已有实例self._instances[interface] instancedef resolve(self, interface):解析依赖# 先检查已注册的实例if interface in self._instances:return self._instances[interface]# 检查工厂if interface not in self._factories:raise KeyError(f未注册的依赖: {interface})factory self._factories[interface]# 单例模式if interface in self._singletons:if interface not in self._instances:self._instances[interface] factory(self)return self._instances[interface]return factory(self)def __getitem__(self, interface):return self.resolve(interface)# 使用container Container()# 注册依赖container.register(Database, lambda c: PostgresDatabase(hostlocalhost), singletonTrue)container.register(EmailSender, lambda c: SMTPMailer(hostsmtp.example.com))container.register(Logger, lambda c: FileLogger(app.log), singletonTrue)container.register(UserService, lambda c: UserService(dbc[Database],emailc[EmailSender],loggerc[Logger]))# 解析user_service container[UserService]user_service.register(Alice, aliceexample.com)四、基于装饰器的DIimport inspectfrom typing import get_type_hintsclass AutoWireContainer:自动装配的DI容器def __init__(self):self._registry {}def register(self, cls, singletonFalse):装饰器注册类def decorator(impl):self._registry[cls] {impl: impl,singleton: singleton,instance: None}return implreturn decoratordef resolve(self, cls):if cls not in self._registry:raise KeyError(f未注册: {cls})entry self._registry[cls]if entry[singleton] and entry[instance]:return entry[instance]# 自动解析构造函数参数impl entry[impl]hints get_type_hints(impl.__init__)kwargs {}for param_name, param_type in hints.items():if param_name return:continueif param_type in self._registry:kwargs[param_name] self.resolve(param_type)instance impl(**kwargs)if entry[singleton]:entry[instance] instancereturn instance# 使用container AutoWireContainer()container.register(Database, singletonTrue)class PostgresDB:def query(self, sql):return []def insert(self, table, data):return datacontainer.register(EmailSender)class SMTPEmail:def send(self, to, subject, body):print(f发送邮件到 {to})container.register(Logger, singletonTrue)class AppLogger:def info(self, msg):print(f[INFO] {msg})def error(self, msg):print(f[ERROR] {msg})container.register(UserService)class UserServiceImpl:def __init__(self, db: Database, email: EmailSender, logger: Logger):self.db dbself.email emailself.logger logger# 自动装配service container.resolve(UserService)五、作用域管理from contextlib import contextmanagerfrom enum import Enumclass Scope(Enum):TRANSIENT transient # 每次创建新实例SINGLETON singleton # 全局单例SCOPED scoped # 作用域内单例class ScopedContainer:def __init__(self, parentNone):self._registry parent._registry if parent else {}self._scoped_instances {}self._singleton_instances parent._singleton_instances if parent else {}def register(self, interface, factory, scopeScope.TRANSIENT):self._registry[interface] {factory: factory, scope: scope}def resolve(self, interface):if interface not in self._registry:raise KeyError(f未注册: {interface})entry self._registry[interface]scope entry[scope]factory entry[factory]if scope Scope.SINGLETON:if interface not in self._singleton_instances:self._singleton_instances[interface] factory(self)return self._singleton_instances[interface]elif scope Scope.SCOPED:if interface not in self._scoped_instances:self._scoped_instances[interface] factory(self)return self._scoped_instances[interface]else: # TRANSIENTreturn factory(self)contextmanagerdef create_scope(self):创建子作用域child ScopedContainer(parentself)try:yield childfinally:# 清理作用域内的资源for instance in child._scoped_instances.values():if hasattr(instance, close):instance.close()# 使用Web请求场景container ScopedContainer()container.register(Database, lambda c: DatabaseConnection(), scopeScope.SCOPED)container.register(UserService, lambda c: UserService(c.resolve(Database)), scopeScope.SCOPED)# 每个请求一个作用域def handle_request(request):with container.create_scope() as scope:service scope.resolve(UserService)# 同一作用域内共享同一个数据库连接return service.process(request)六、依赖注入与测试class TestUserService:def setup_method(self):每个测试方法前创建Mock依赖self.mock_db MockDatabase()self.mock_email MockEmailSender()self.mock_logger MockLogger()self.service UserService(dbself.mock_db,emailself.mock_email,loggerself.mock_logger)def test_register_success(self):self.mock_db.insert_returns {id: 1, name: Alice}result self.service.register(Alice, aliceexample.com)assert result[name] Aliceassert self.mock_email.sent_count 1assert self.mock_logger.messages[-1] 注册用户: Alicedef test_register_db_failure(self):self.mock_db.should_fail Truewith pytest.raises(DatabaseError):self.service.register(Alice, aliceexample.com)assert self.mock_email.sent_count 0 # 邮件未发送class MockDatabase:def __init__(self):self.insert_returns {}self.should_fail Falseself.queries []def insert(self, table, data):if self.should_fail:raise DatabaseError(模拟数据库错误)self.queries.append((insert, table, data))return self.insert_returnsdef query(self, sql):self.queries.append((query, sql))return []七、FastAPI的依赖注入from fastapi import FastAPI, Dependsapp FastAPI()# 依赖函数def get_db():db DatabaseSession()try:yield dbfinally:db.close()def get_current_user(token: str Depends(get_token)):user verify_token(token)if not user:raise HTTPException(status_code401)return userdef get_user_service(db: Database Depends(get_db),current_user: User Depends(get_current_user)):return UserService(dbdb, usercurrent_user)# 路由中使用app.post(/orders)async def create_order(order_data: OrderCreate,service: UserService Depends(get_user_service)):return service.create_order(order_data)# 依赖覆盖测试时def override_get_db():return FakeDatabase()app.dependency_overrides[get_db] override_get_db八、实际项目中的DI架构# 项目结构# src/# interfaces/ # 抽象接口定义# services/ # 业务逻辑# repositories/ # 数据访问# infrastructure/ # 具体实现# container.py # DI配置# container.pydef create_container(config):container Container()# 基础设施container.register(Database,lambda c: PostgresDatabase(config.db_url), singletonTrue)container.register(Cache,lambda c: RedisCache(config.redis_url), singletonTrue)container.register(EmailSender,lambda c: SMTPMailer(config.smtp_host))# 仓储层container.register(UserRepository,lambda c: SQLUserRepository(c[Database]))container.register(OrderRepository,lambda c: SQLOrderRepository(c[Database]))# 服务层container.register(UserService,lambda c: UserService(c[UserRepository], c[EmailSender]))container.register(OrderService,lambda c: OrderService(c[OrderRepository], c[UserService], c[Cache]))return container九、总结依赖注入的好处1. 松耦合组件不依赖具体实现2. 可测试轻松替换为Mock对象3. 可配置运行时决定使用哪个实现4. 单一职责每个类只关注自己的逻辑使用建议- 小项目手动构造函数注入即可- 中型项目简单的DI容器- 大型项目使用成熟的DI框架dependency-injector- Web框架利用框架内置的DIFastAPI Depends- 不要过度设计只在确实需要灵活性时使用DI