Python依赖注入模式

依赖注入（DI）是实现控制反转的核心技术，提高代码可测试性和可维护性。

依赖注入基础

紧耦合问题

# 紧耦合：直接在类内部创建依赖
class UserService:
    def __init__(self):
        self.db = MySQLDatabase()        # 紧耦合
        self.logger = FileLogger()       # 紧耦合
        self.email = SMTPClient()        # 紧耦合

    def create_user(self, name, email):
        user = User(name, email)
        self.db.save(user)
        self.logger.log(f"Created {name}")
        self.email.send(email, "Welcome!")

# 问题：
# 1. 无法切换数据库
# 2. 测试困难（无法mock）
# 3. 配置变化需要修改代码

依赖注入解耦

# 松耦合：依赖从外部注入
class UserService:
    def __init__(self, db, logger, email):
        self.db = db      # 注入依赖
        self.logger = logger
        self.email = email

    def create_user(self, name, email):
        user = User(name, email)
        self.db.save(user)
        self.logger.log(f"Created {name}")
        self.email.send(email, "Welcome!")

# 好处：
# 1. 可替换任何实现
# 2. 测试时注入mock对象
# 3. 配置与代码分离

注入方式

构造器注入（推荐）

from abc import ABC, abstractmethod

# 定义抽象接口
class Database(ABC):
    @abstractmethod
    def save(self, entity):
        pass

class Logger(ABC):
    @abstractmethod
    def log(self, message):
        pass

class EmailSender(ABC):
    @abstractmethod
    def send(self, to, message):
        pass

# 通过构造器注入依赖
class UserService:
    def __init__(self, db: Database, logger: Logger, email: EmailSender):
        self.db = db
        self.logger = logger
        self.email = email

    def create_user(self, name, email):
        user = User(name, email)
        self.db.save(user)
        self.logger.log(f"Created {name}")
        self.email.send(email, "Welcome!")

# 使用
service = UserService(
    MySQLDatabase(),
    FileLogger(),
    SMTPClient()
)

属性注入

class UserService:
    def __init__(self):
        self.db = None
        self.logger = None
        self.email = None

    def create_user(self, name, email):
        if self.db is None:
            raise RuntimeError("Database not configured")

        user = User(name, email)
        self.db.save(user)
        self.logger.log(f"Created {name}")
        self.email.send(email, "Welcome!")

# 属性注入（后期配置）
service = UserService()
service.db = MySQLDatabase()
service.logger = FileLogger()
service.email = SMTPClient()

方法注入

class UserService:
    def create_user(self, name, email, db=None, logger=None):
        db = db or self.default_db
        logger = logger or self.default_logger

        user = User(name, email)
        db.save(user)
        logger.log(f"Created {name}")

        return user

# 每次调用指定不同依赖
service.create_user("Alice", "a@b.com", db=test_db)

依赖注入容器

简单容器实现

class DIContainer:
    def __init__(self):
        self._services = {}
        self._instances = {}

    def register(self, name, factory):
        "注册服务工厂"
        self._services[name] = factory

    def get(self, name):
        "获取服务实例"
        if name not in self._instances:
            factory = self._services[name]
            self._instances[name] = factory(self)
        return self._instances[name]

    def get_new(self, name):
        "每次创建新实例"
        return self._services[name](self)

# 使用
container = DIContainer()

# 注册服务
container.register('db', lambda c: MySQLDatabase('localhost'))
container.register('logger', lambda c: FileLogger('app.log'))
container.register('email', lambda c: SMTPClient())

container.register('user_service', lambda c: UserService(
    c.get('db'),
    c.get('logger'),
    c.get('email')
))

# 获取服务
service = container.get('user_service')

支持接口注册

class Container:
    def __init__(self):
        self._bindings = {}
        self._instances = {}

    def bind(self, interface, implementation, singleton=True):
        "绑定接口到实现"
        self._bindings[interface] = {
            'implementation': implementation,
            'singleton': singleton
        }

    def resolve(self, interface):
        "解析接口获取实例"
        binding = self._bindings.get(interface)
        if not binding:
            raise KeyError(f"No binding for {interface}")

        if binding['singleton']:
            if interface not in self._instances:
                self._instances[interface] = binding['implementation]()
            return self._instances[interface]

        return binding['implementation']()

# 使用
container = Container()
container.bind(Database, MySQLDatabase, singleton=True)
container.bind(Logger, FileLogger, singleton=True)
container.bind(EmailSender, SMTPClient)

db = container.resolve(Database)

自动注入

import inspect

class AutoContainer:
    def __init__(self):
        self._bindings = {}

    def bind(self, interface, implementation):
        self._bindings[interface] = implementation

    def resolve(self, interface):
        impl = self._bindings.get(interface)
        if not impl:
            raise KeyError(f"No binding for {interface}")

        # 分析构造器参数
        sig = inspect.signature(impl.__init__)
        kwargs = {}

        for param_name, param in sig.parameters.items():
            if param_name == 'self':
                continue

            # 根据参数类型自动注入
            if param.annotation != inspect.Parameter.empty:
                kwargs[param_name] = self.resolve(param.annotation)

        return impl(**kwargs)

# 使用
class UserService:
    def __init__(self, db: Database, logger: Logger):
        self.db = db
        self.logger = logger

container = AutoContainer()
container.bind(Database, MySQLDatabase)
container.bind(Logger, FileLogger)

# 自动注入构造器参数
service = container.resolve(UserService)

使用第三方库

dependency-injector库

pip install dependency-injector

from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    database = providers.Singleton(
        MySQLDatabase,
        connection_string=config.db.connection_string
    )

    logger = providers.Singleton(
        FileLogger,
        filename=config.log.filename
    )

    email_sender = providers.Singleton(
        SMTPClient,
        host=config.email.host
    )

    user_service = providers.Factory(
        UserService,
        db=database,
        logger=logger,
        email=email_sender
    )

# 配置
container = Container()
container.config.from_dict({
    'db': {'connection_string': 'mysql://...'},
    'log': {'filename': 'app.log'},
    'email': {'host': 'smtp.example.com'}
})

# 获取服务
service = container.user_service()

FastAPI依赖注入

from fastapi import Depends, FastAPI

app = FastAPI()

# 定义依赖
def get_db():
    db = Database()
    try:
        yield db
    finally:
        db.close()

def get_current_user(token: str = Depends(get_token)):
    return verify_token(token)

# 使用依赖
@app.get("/users/me")
def get_me(user: User = Depends(get_current_user), db: Database = Depends(get_db)):
    return db.query(User).get(user.id)

测试中的应用

Mock注入测试

import unittest
from unittest.mock import Mock

class UserServiceTest(unittest.TestCase):
    def setUp(self):
        # 创建mock对象
        self.mock_db = Mock(spec=Database)
        self.mock_logger = Mock(spec=Logger)
        self.mock_email = Mock(spec=EmailSender)

        # 注入mock
        self.service = UserService(
            self.mock_db,
            self.mock_logger,
            self.mock_email
        )

    def test_create_user(self):
        # 配置mock行为
        self.mock_db.save.return_value = User(id=1, name='Test')

        # 执行测试
        result = self.service.create_user('Alice', 'alice@test.com')

        # 验证调用
        self.mock_db.save.assert_called_once()
        self.mock_logger.log.assert_called_with('Created Alice')
        self.mock_email.send.assert_called_with('alice@test.com', 'Welcome!')

        self.assertEqual(result.id, 1)

测试容器

class TestContainer(Container):
    "测试专用容器"
    def __init__(self):
        super().__init__()
        self.bind(Database, MockDatabase)  # 测试数据库
        self.bind(Logger, MockLogger)      # 测试日志
        self.bind(EmailSender, MockEmail)  # 测试邮件

def test_with_container():
    container = TestContainer()
    service = container.resolve(UserService)

    user = service.create_user('Test', 'test@test.com')
    assert user is not None

依赖注入最佳实践

设计原则

# 1. 依赖抽象而非具体
class UserService:
    def __init__(self, db: Database):  # 接口类型
        pass

# 2. 避免过度注入
class GoodService:
    def __init__(self, db: Database):  # 只注入必要依赖
        pass

class BadService:
    def __init__(self, db, logger, email, cache, queue, config):  # 过多依赖
        pass

# 3. 使用服务定位器模式（备选）
class ServiceLocator:
    _services = {}

    @classmethod
    def register(cls, name, service):
        cls._services[name] = service

    @classmethod
    def get(cls, name):
        return cls._services[name]

配置管理

class Config:
    def __init__(self, env_file='.env'):
        self._load_config(env_file)

    def _load_config(self, file):
        # 从环境变量或配置文件加载
        self.db_url = os.getenv('DB_URL', 'sqlite:///default.db')
        self.log_file = os.getenv('LOG_FILE', 'app.log')
        self.smtp_host = os.getenv('SMTP_HOST', 'localhost')

# 容器配置
container = Container()
config = Config()

container.register('db', lambda c: Database(c.get('config').db_url))
container.register('config', lambda c: config)

注入方式对比

注意：依赖注入要适度，过多依赖注入会增加复杂度，小型应用可以简化。

要点总结

• 构造器注入：依赖从外部传入，明确必需依赖，推荐默认方式
• 依赖注入容器：统一管理服务注册和获取，支持单例和多例
• 自动注入：通过反射分析构造器参数，自动解析依赖
• 测试友好：注入mock对象替代真实实现，隔离测试
• FastAPI内置DI：使用Depends实现请求级依赖注入
• 依赖抽象：注入接口而非具体实现，实现可替换和可测试

存放路径：articles/PYTHON/专家/架构与设计/依赖注入模式.md