一、前言

作为从 Java 转向 Python 的开发者，你可能会困惑：Python 中该如何组织代码？什么时候用模块，什么时候用类？单例模式怎么实现？依赖注入又该如何处理？

本文将系统性地解答这些问题，帮助你在 Python 项目中做出正确的架构设计决策。

二、模块 vs 类：如何选择？

2.1、Python 模块的本质

在 Python 中，模块本身就是单例。这是 Python 与 Java 的重要区别之一。

# config.py
database_url: str = "postgresql://localhost/mydb"
max_connections: int = 10

def get_connection_string() -> str:
    return f"{database_url}?max_connections={max_connections}"

# main.py
import config

print(config.database_url)  # 直接使用模块级变量
print(config.get_connection_string())

模块的特点：

• 模块在首次导入时被加载，之后的导入都返回同一个模块对象
• 模块级变量就是全局单例
• 适合无状态或共享状态的场景

2.2、类的使用场景

类适合需要封装状态和多实例的场景。

class DatabaseConnection:
    """数据库连接类，每个实例维护独立的连接状态"""
    
    def __init__(self, url: str, max_connections: int = 10):
        self.url: str = url
        self.max_connections: int = max_connections
        self._connection = None
    
    def connect(self) -> None:
        """建立数据库连接"""
        # 连接逻辑
        pass
    
    def close(self) -> None:
        """关闭连接"""
        if self._connection:
            self._connection.close()

类的特点：

• 可以创建多个实例，每个实例有独立状态
• 支持继承和多态
• 更符合面向对象的设计思想

2.3、选择决策树

需要多个实例？
├─ 是 → 使用类
└─ 否
    ├─ 需要复杂的状态管理？
    │   ├─ 是 → 使用类（单例模式）
    │   └─ 否 → 使用模块
    └─ 只需要工具函数？
        └─ 使用模块

实践建议：

• 配置信息：优先使用模块（如 config.py）
• 工具函数：使用模块（如 utils.py、helpers.py）
• 业务逻辑：优先使用类（如 UserService、OrderManager）
• 数据模型：使用类（如 User、Product）

三、单例模式深度解析

3.1、模块级单例 vs 类单例模式

在 Python 中，模块本身就是天然的单例，这与 Java 有本质区别。

3.1.1、模块单例的原理

# database.py
class _DatabaseConnection:
    """私有数据库连接类"""
    
    def __init__(self):
        self.url: str = ""
        self._connection = None
        print("Database instance created")
    
    def connect(self, url: str) -> None:
        """连接数据库"""
        self.url = url
        print(f"Connected to {url}")
    
    def query(self, sql: str) -> list:
        """执行查询"""
        return []

db: _DatabaseConnection = _DatabaseConnection()

关键特性：

1. 模块在整个 Python 进程中只会加载一次
2. 模块级变量 db 在首次 import 时创建，后续 import 都返回同一个对象
3. 线程安全由 Python 解释器保证（GIL + import lock）

验证：

# module_a.py
from database import db
db.connect("postgresql://localhost/db1")
print(f"module_a: {id(db)}")

# module_b.py
from database import db
print(f"module_b: {id(db)}")  # 与 module_a 中的 id 相同
print(f"URL: {db.url}")  # 输出: postgresql://localhost/db1

3.1.2、类单例模式的动机

既然模块已经是单例，为什么还需要类单例模式？

主要场景：

1. 延迟初始化：模块导入时不立即创建对象，而是在首次使用时创建
2. 参数化构造：首次创建时需要传入参数
3. 继承需求：需要让子类也具有单例特性
4. 动态切换：在测试或特殊场景下需要重置单例

对比示例：

# 模块单例：导入时立即初始化
# config.py
class Config:
    def __init__(self):
        self.database_url = "postgresql://localhost/db"

config = Config()  # 模块导入时立即创建

# 类单例：延迟初始化，支持参数
# database.py
class Database:
    _instance = None
    
    def __new__(cls, url: str = ""):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.url = url  # 保存首次传入的参数
            print(f"Database created with {url}")
        return cls._instance

# 使用
db1 = Database("postgresql://localhost/db1")  # 输出: Database created with postgresql://localhost/db1
db2 = Database("postgresql://localhost/db2")  # 不会再次创建
print(db1.url)  # 输出: postgresql://localhost/db1（保持首次参数）
print(db1 is db2)  # True

3.2、线程安全的单例模式

Python 的 GIL（全局解释器锁）并不能完全保证单例的线程安全，尤其是在类单例模式中。

3.2.1、非线程安全的单例（错误示范）

import time
from threading import Thread

class UnsafeSingleton:
    """非线程安全的单例"""
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            time.sleep(0.1)  # 模拟耗时操作
            cls._instance = super().__new__(cls)
            print(f"Instance created: {id(cls._instance)}")
        return cls._instance

def create_singleton():
    """创建单例"""
    singleton = UnsafeSingleton()

threads = [Thread(target=create_singleton) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

可能输出（创建了多个实例）：

Instance created: 140234567890123
Instance created: 140234567890456
Instance created: 140234567890789

3.2.2、使用锁保证线程安全

import threading
from typing import Optional

class ThreadSafeSingleton:
    """线程安全的单例"""
    _instance: Optional['ThreadSafeSingleton'] = None
    _lock: threading.Lock = threading.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:  # 双重检查
                    cls._instance = super().__new__(cls)
                    print(f"Instance created: {id(cls._instance)}")
        return cls._instance

def create_singleton():
    """创建单例"""
    singleton = ThreadSafeSingleton()

threads = [threading.Thread(target=create_singleton) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

输出（只创建一次）：

Instance created: 140234567890123

**双重检查锁定（Double-Checked Locking）**的关键：

1. 第一次检查：避免每次都获取锁，提高性能
2. 加锁：保证只有一个线程进入临界区
3. 第二次检查：防止多个线程同时通过第一次检查后创建多个实例

3.2.3、使用元类实现线程安全单例

import threading
from typing import Any

class ThreadSafeSingletonMeta(type):
    """线程安全的单例元类"""
    
    _instances: dict[type, object] = {}
    _lock: threading.Lock = threading.Lock()
    
    def __call__(cls, *args: Any, **kwargs: Any) -> Any:
        if cls not in cls._instances:
            with cls._lock:
                if cls not in cls._instances:
                    instance = super().__call__(*args, **kwargs)
                    cls._instances[cls] = instance
                    print(f"{cls.__name__} instance created")
        return cls._instances[cls]

class Database(metaclass=ThreadSafeSingletonMeta):
    """数据库类"""
    
    def __init__(self, url: str = ""):
        self.url = url

def create_db():
    """创建数据库实例"""
    db = Database("postgresql://localhost/mydb")

threads = [threading.Thread(target=create_db) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

3.3、装饰器单例（推荐）

最优雅的线程安全单例实现方式。

import threading
from typing import Callable, TypeVar, cast, Any
from functools import wraps

T = TypeVar('T')

def singleton(cls: type[T]) -> Callable[..., T]:
    """线程安全的单例装饰器"""
    instances: dict[type, object] = {}
    lock = threading.Lock()
    
    @wraps(cls)
    def get_instance(*args: Any, **kwargs: Any) -> T:
        if cls not in instances:
            with lock:
                if cls not in instances:
                    instances[cls] = cls(*args, **kwargs)
                    print(f"{cls.__name__} instance created")
        return cast(T, instances[cls])
    
    return get_instance

@singleton
class CacheManager:
    """缓存管理器"""
    
    def __init__(self):
        self._cache: dict[str, Any] = {}
    
    def set(self, key: str, value: Any) -> None:
        """设置缓存"""
        self._cache[key] = value
    
    def get(self, key: str) -> Any:
        """获取缓存"""
        return self._cache.get(key)

cache1 = CacheManager()
cache2 = CacheManager()
print(cache1 is cache2)  # True

3.4、模块级单例（最推荐）

模块级单例是天然线程安全的，因为：

1. Python 的 import 机制使用了内部锁
2. 模块只会初始化一次
3. 简单、直观、Pythonic

# database.py
import threading

class _DatabaseConnection:
    """私有数据库连接类"""
    
    def __init__(self):
        self.url: str = ""
        self._connection = None
        self._lock = threading.Lock()
        print("Database instance created")
    
    def connect(self, url: str) -> None:
        """连接数据库（线程安全）"""
        with self._lock:
            self.url = url
            print(f"Connected to {url}")
    
    def query(self, sql: str) -> list:
        """执行查询"""
        return []

db: _DatabaseConnection = _DatabaseConnection()

优点：

• 导入时自动初始化，天然线程安全
• 代码简洁，易于理解
• 易于测试（可以 mock 整个模块）
• 符合 Python 风格

3.5、单例模式对比总结

实现方式	线程安全	延迟初始化	参数化构造	复杂度	推荐场景
模块级单例	✅	❌	❌	⭐	首选方案
装饰器单例	✅	✅	⚠️	⭐⭐	需要延迟初始化
元类单例	✅	✅	✅	⭐⭐⭐	需要继承单例特性
__new__ 单例	⚠️	✅	✅	⭐⭐	需要手动加锁

选择建议：

• 90% 的场景：使用模块级单例
• 需要延迟初始化：使用装饰器单例
• 需要单例继承：使用元类单例
• 避免使用：__new__ 单例（需要手动处理线程安全）

四、为什么需要依赖注入？

4.1、从全局单例的痛点说起

假设你正在开发一个用户管理系统，自然而然地使用了全局单例：

# database.py
class Database:
    """数据库类"""
    
    def __init__(self):
        self.connection = None
    
    def connect(self, url: str) -> None:
        """连接数据库"""
        print(f"Connecting to {url}")
    
    def query(self, sql: str) -> list:
        """执行查询"""
        print(f"Executing: {sql}")
        return [{"id": 1, "name": "Alice"}]

db = Database()
db.connect("postgresql://localhost/mydb")

# user_service.py
from database import db

class UserService:
    """用户服务"""
    
    def get_user(self, user_id: int) -> dict:
        """获取用户"""
        result = db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}
    
    def create_user(self, name: str) -> dict:
        """创建用户"""
        db.query(f"INSERT INTO users (name) VALUES ('{name}')")
        return {"name": name}

看起来很简单，但问题随之而来：

4.1.1、测试难题

当你想测试 UserService 时，问题出现了：

# test_user_service.py
import pytest
from user_service import UserService

def test_get_user():
    """测试获取用户"""
    service = UserService()
    user = service.get_user(1)
    
    # 问题：这会真的连接数据库！
    # - 测试依赖真实数据库
    # - 测试速度慢
    # - 无法控制返回数据
    # - 可能影响生产数据
    assert user["name"] == "Alice"

你无法 mock 数据库，因为 UserService 硬编码依赖 database.db。

4.1.2、灵活性问题

现在需求变了，需要支持多个数据库：

# 新需求：主库用于写，从库用于读
class UserService:
    def get_user(self, user_id: int) -> dict:
        """从只读库读取"""
        # 问题：怎么切换到从库？
        result = db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}
    
    def create_user(self, name: str) -> dict:
        """写入主库"""
        # 需要连接主库
        # 但 db 是全局的，无法灵活切换
        db.query(f"INSERT INTO users (name) VALUES ('{name}')")
        return {"name": name}

全局单例导致无法灵活切换不同的数据库实例。

4.1.3、耦合度高

UserService 与具体的 Database 实现强耦合：

# 如果需要切换到 MongoDB，必须修改 UserService 的代码
from database import db  # 强依赖 PostgreSQL 实现

class UserService:
    def get_user(self, user_id: int) -> dict:
        # 直接使用 db，无法切换到 MongoDB
        result = db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}

违反了开闭原则（对扩展开放，对修改封闭）。

4.2、依赖注入：解决方案

依赖注入（Dependency Injection，DI）的核心思想：不要自己创建依赖，而是由外部传入。

4.2.1、基础示例

from typing import Protocol

class DatabaseProtocol(Protocol):
    """数据库接口（协议）"""
    
    def query(self, sql: str) -> list:
        """执行查询"""
        ...

class UserService:
    """用户服务"""
    
    def __init__(self, database: DatabaseProtocol):
        """通过构造函数注入依赖"""
        self.db = database
    
    def get_user(self, user_id: int) -> dict:
        """获取用户"""
        result = self.db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}

# 使用
from database import db

user_service = UserService(database=db)  # 外部传入依赖
user = user_service.get_user(1)

4.2.2、解决测试问题

现在可以轻松 mock 数据库：

# test_user_service.py
from unittest.mock import Mock
from user_service import UserService

def test_get_user():
    """测试获取用户"""
    # 创建 mock 数据库
    mock_db = Mock()
    mock_db.query.return_value = [{"id": 1, "name": "Bob"}]
    
    # 注入 mock 依赖
    service = UserService(database=mock_db)
    
    # 执行测试
    user = service.get_user(1)
    
    # 验证
    assert user["name"] == "Bob"
    mock_db.query.assert_called_once_with("SELECT * FROM users WHERE id = 1")

优点：

• 不依赖真实数据库
• 测试速度快
• 可以控制返回数据
• 可以验证调用行为

4.2.3、解决灵活性问题

可以轻松切换不同的数据库实例：

class Database:
    """PostgreSQL 数据库"""
    
    def __init__(self, url: str):
        self.url = url
    
    def query(self, sql: str) -> list:
        print(f"Querying {self.url}: {sql}")
        return []

# 创建主库和从库
master_db = Database("postgresql://master/mydb")
slave_db = Database("postgresql://slave/mydb")

# 根据场景注入不同的依赖
read_service = UserService(database=slave_db)  # 读操作用从库
write_service = UserService(database=master_db)  # 写操作用主库

4.2.4、解决耦合问题

通过接口（Protocol）编程，轻松切换实现：

from typing import Protocol

class DatabaseProtocol(Protocol):
    """数据库接口"""
    
    def query(self, sql: str) -> list:
        ...

class PostgreSQLDatabase:
    """PostgreSQL 实现"""
    
    def query(self, sql: str) -> list:
        print(f"PostgreSQL: {sql}")
        return []

class MongoDatabase:
    """MongoDB 实现"""
    
    def query(self, sql: str) -> list:
        print(f"MongoDB: {sql}")
        return []

class UserService:
    """用户服务（依赖接口，而非具体实现）"""
    
    def __init__(self, database: DatabaseProtocol):
        self.db = database
    
    def get_user(self, user_id: int) -> dict:
        result = self.db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}

# 可以自由切换实现
pg_service = UserService(database=PostgreSQLDatabase())
mongo_service = UserService(database=MongoDatabase())

4.3、依赖注入的三种方式

4.3.1、构造函数注入（推荐）

class UserService:
    """用户服务"""
    
    def __init__(self, database: DatabaseProtocol, cache: CacheProtocol):
        """通过构造函数注入所有依赖"""
        self.db = database
        self.cache = cache

优点：

• 依赖关系清晰
• 对象创建后依赖就绪
• 易于测试

4.3.2、属性注入

class UserService:
    """用户服务"""
    
    def __init__(self):
        self.db: Optional[DatabaseProtocol] = None
        self.cache: Optional[CacheProtocol] = None

# 使用
service = UserService()
service.db = database  # 通过属性注入
service.cache = cache

缺点：

• 依赖可能未初始化
• 增加了空值检查的负担

4.3.3、方法注入

class UserService:
    """用户服务"""
    
    def get_user(self, user_id: int, database: DatabaseProtocol) -> dict:
        """通过方法参数注入"""
        result = database.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}

适用场景：

• 依赖只在特定方法中使用
• 不同调用需要不同依赖

4.4、依赖注入 vs 全局单例对比

维度	全局单例	依赖注入
测试难度	❌ 难以 mock	✅ 易于 mock
灵活性	❌ 固定实现	✅ 可切换实现
耦合度	❌ 高耦合	✅ 低耦合
代码复杂度	✅ 简单	⚠️ 稍复杂
适用场景	简单项目、配置	中大型项目、业务逻辑

选择建议：

• 配置信息、工具类：使用全局单例（模块级）
• 业务逻辑、服务类：使用依赖注入
• 数据访问层：使用依赖注入

五、依赖注入容器

5.1、为什么需要 DI 容器？

当项目变大，依赖关系变复杂时，手动管理依赖会很繁琐。

5.1.1、依赖管理的痛点

# 手动管理依赖
database = Database("postgresql://localhost/mydb")
cache = RedisCache("redis://localhost:6379")
logger = Logger("app.log")

user_repository = UserRepository(database)
order_repository = OrderRepository(database)

user_service = UserService(user_repository, cache, logger)
order_service = OrderService(order_repository, user_service, cache, logger)

# 问题：
# 1. 依赖创建顺序必须手动管理
# 2. 代码重复（每个地方都要创建依赖）
# 3. 单例管理困难（如何保证 database 只创建一次？）
# 4. 配置分散（URL、端口等配置散落各处）

5.1.2、DI 容器的价值

依赖注入容器（DI Container）就像一个"依赖工厂"，统一管理依赖的创建、生命周期和注入。

核心功能：

1. 服务注册：告诉容器如何创建对象
2. 依赖解析：自动创建对象及其依赖
3. 生命周期管理：单例、瞬态等
4. 配置集中：统一管理配置

5.2、简单的 DI 容器实现

5.2.1、基础版本

from typing import TypeVar, Callable, Any, Dict

T = TypeVar('T')

class DIContainer:
    """简单的依赖注入容器"""
    
    def __init__(self):
        self._factories: Dict[type, Callable[[], Any]] = {}
        self._singletons: Dict[type, Any] = {}
    
    def register(self, service_type: type[T], factory: Callable[[], T], singleton: bool = True) -> None:
        """注册服务
        
        Args:
            service_type: 服务类型
            factory: 工厂函数
            singleton: 是否单例
        """
        self._factories[service_type] = factory
        if singleton:
            self._singletons[service_type] = None
    
    def resolve(self, service_type: type[T]) -> T:
        """解析服务
        
        Args:
            service_type: 服务类型
            
        Returns:
            服务实例
        """
        if service_type in self._singletons:
            if self._singletons[service_type] is None:
                self._singletons[service_type] = self._factories[service_type]()
            return self._singletons[service_type]
        
        if service_type in self._factories:
            return self._factories[service_type]()
        
        raise ValueError(f"Service {service_type} not registered")

# 使用示例
class Database:
    """数据库类"""
    
    def __init__(self, url: str):
        self.url = url
        print(f"Database created: {url}")

class UserRepository:
    """用户仓储"""
    
    def __init__(self, database: Database):
        self.db = database
        print("UserRepository created")

class UserService:
    """用户服务"""
    
    def __init__(self, repository: UserRepository):
        self.repository = repository
        print("UserService created")

# 创建容器并注册服务
container = DIContainer()

container.register(
    Database,
    lambda: Database("postgresql://localhost/mydb"),
    singleton=True
)

container.register(
    UserRepository,
    lambda: UserRepository(container.resolve(Database)),
    singleton=True
)

container.register(
    UserService,
    lambda: UserService(container.resolve(UserRepository)),
    singleton=False  # 每次创建新实例
)

# 使用
service1 = container.resolve(UserService)  # 创建所有依赖
service2 = container.resolve(UserService)  # 复用单例依赖

print(service1.repository.db is service2.repository.db)  # True（单例）
print(service1 is service2)  # False（瞬态）

输出：

Database created: postgresql://localhost/mydb
UserRepository created
UserService created
UserService created
True
False

5.2.2、支持自动解析的版本

import inspect
from typing import TypeVar, Callable, Any, Dict, get_type_hints

T = TypeVar('T')

class AutoDIContainer:
    """支持自动依赖解析的 DI 容器"""
    
    def __init__(self):
        self._factories: Dict[type, Callable[[], Any]] = {}
        self._singletons: Dict[type, Any] = {}
    
    def register(self, service_type: type[T], implementation: type[T] = None, singleton: bool = True) -> None:
        """注册服务（自动解析构造函数依赖）
        
        Args:
            service_type: 服务类型
            implementation: 实现类型（默认与 service_type 相同）
            singleton: 是否单例
        """
        impl = implementation or service_type
        
        def factory() -> T:
            # 获取构造函数的类型提示
            sig = inspect.signature(impl.__init__)
            params = sig.parameters
            
            # 解析依赖
            kwargs = {}
            for name, param in params.items():
                if name == 'self':
                    continue
                if param.annotation != inspect.Parameter.empty:
                    kwargs[name] = self.resolve(param.annotation)
            
            return impl(**kwargs)
        
        self._factories[service_type] = factory
        if singleton:
            self._singletons[service_type] = None
    
    def resolve(self, service_type: type[T]) -> T:
        """解析服务"""
        if service_type in self._singletons:
            if self._singletons[service_type] is None:
                self._singletons[service_type] = self._factories[service_type]()
            return self._singletons[service_type]
        
        if service_type in self._factories:
            return self._factories[service_type]()
        
        raise ValueError(f"Service {service_type} not registered")

# 使用示例（无需手动指定依赖关系）
class Config:
    """配置类"""
    
    def __init__(self):
        self.db_url = "postgresql://localhost/mydb"

class Database:
    """数据库类"""
    
    def __init__(self, config: Config):
        self.url = config.db_url
        print(f"Database created: {self.url}")

class UserRepository:
    """用户仓储"""
    
    def __init__(self, database: Database):
        self.db = database
        print("UserRepository created")

class UserService:
    """用户服务"""
    
    def __init__(self, repository: UserRepository):
        self.repository = repository
        print("UserService created")

# 创建容器并注册服务（自动解析依赖）
container = AutoDIContainer()

container.register(Config, singleton=True)
container.register(Database, singleton=True)
container.register(UserRepository, singleton=True)
container.register(UserService, singleton=False)

# 使用（容器自动解析依赖链）
service = container.resolve(UserService)

输出：

Database created: postgresql://localhost/mydb
UserRepository created
UserService created

5.3、使用成熟的 DI 库

5.3.1、dependency-injector（推荐）

pip install dependency-injector

基础用法：

from dependency_injector import containers, providers

class Database:
    """数据库类"""
    
    def __init__(self, url: str):
        self.url = url
        print(f"Database created: {url}")
    
    def query(self, sql: str) -> list:
        """执行查询"""
        return []

class CacheService:
    """缓存服务"""
    
    def __init__(self, redis_url: str):
        self.redis_url = redis_url
        print(f"Cache created: {redis_url}")

class UserRepository:
    """用户仓储"""
    
    def __init__(self, database: Database):
        self.db = database
        print("UserRepository created")

class UserService:
    """用户服务"""
    
    def __init__(self, repository: UserRepository, cache: CacheService):
        self.repository = repository
        self.cache = cache
        print("UserService created")

class Container(containers.DeclarativeContainer):
    """依赖容器"""
    
    config = providers.Configuration()
    
    database = providers.Singleton(
        Database,
        url=config.database.url,
    )
    
    cache = providers.Singleton(
        CacheService,
        redis_url=config.cache.url,
    )
    
    user_repository = providers.Singleton(
        UserRepository,
        database=database,
    )
    
    user_service = providers.Factory(
        UserService,
        repository=user_repository,
        cache=cache,
    )

# 使用
container = Container()
container.config.database.url.from_value("postgresql://localhost/mydb")
container.config.cache.url.from_value("redis://localhost:6379")

service = container.user_service()

从配置文件加载：

# config.yml
database:
  url: "postgresql://localhost/mydb"
cache:
  url: "redis://localhost:6379"

container = Container()
container.config.from_yaml("config.yml")

service = container.user_service()

5.3.2、injector

pip install injector

from injector import Injector, inject, singleton

class Database:
    """数据库类"""
    
    @inject
    def __init__(self):
        print("Database created")

class UserRepository:
    """用户仓储"""
    
    @inject
    def __init__(self, database: Database):
        self.db = database
        print("UserRepository created")

class UserService:
    """用户服务"""
    
    @inject
    def __init__(self, repository: UserRepository):
        self.repository = repository
        print("UserService created")

# 使用
injector = Injector()
service = injector.get(UserService)  # 自动解析依赖

5.4、FastAPI 的依赖注入

FastAPI 内置了优雅的依赖注入系统，无需额外的 DI 容器。

from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()

class Database:
    """数据库类"""
    
    def __init__(self):
        print("Database created")
    
    def query(self, sql: str) -> list:
        """执行查询"""
        return [{"id": 1, "name": "Alice"}]

class UserService:
    """用户服务"""
    
    def __init__(self, database: Database):
        self.db = database
    
    def get_user(self, user_id: int) -> dict:
        """获取用户"""
        result = self.db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}

# 依赖提供者
def get_database() -> Database:
    """获取数据库实例（单例）"""
    return Database()

def get_user_service(database: Annotated[Database, Depends(get_database)]) -> UserService:
    """获取用户服务实例"""
    return UserService(database)

# API 路由
@app.get("/users/{user_id}")
async def read_user(
    user_id: int,
    user_service: Annotated[UserService, Depends(get_user_service)]
) -> dict:
    """获取用户接口"""
    return user_service.get_user(user_id)

生命周期管理：

from contextlib import asynccontextmanager

class Database:
    """数据库类"""
    
    def __init__(self):
        self.connection = None
    
    def connect(self) -> None:
        """连接数据库"""
        print("Database connected")
        self.connection = "connection"
    
    def close(self) -> None:
        """关闭连接"""
        print("Database closed")
        self.connection = None

# 应用生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用启动和关闭时的生命周期管理"""
    # 启动时
    db = Database()
    db.connect()
    app.state.db = db
    
    yield
    
    # 关闭时
    db.close()

app = FastAPI(lifespan=lifespan)

# 依赖提供者
async def get_database(request: Request) -> Database:
    """从应用状态获取数据库实例"""
    return request.app.state.db

@app.get("/users/{user_id}")
async def read_user(
    user_id: int,
    database: Annotated[Database, Depends(get_database)]
) -> dict:
    """获取用户接口"""
    return {"id": user_id, "db": database.connection}

5.5、DI 容器选择指南

DI 方案	复杂度	功能	推荐场景
手动注入	⭐	⭐	小型项目
自定义容器	⭐⭐	⭐⭐	学习 DI 原理
dependency-injector	⭐⭐⭐	⭐⭐⭐⭐	通用项目
injector	⭐⭐	⭐⭐⭐	简单 DI 需求
FastAPI Depends	⭐⭐	⭐⭐⭐⭐	Web 应用

选择建议：

• 小型项目：手动注入或模块级单例
• 通用 Python 项目：dependency-injector
• FastAPI 项目：使用 FastAPI 内置的 Depends
• 学习目的：先自己实现简单的容器，再使用成熟库

六、整合：模块、类与依赖管理的架构实践

6.1、实战案例：构建用户管理系统

现在，我们将所有概念整合到一个实际项目中，展示如何在真实场景中应用模块、类、单例和依赖注入。

6.1.1、项目结构设计

user_management/
├── config/
│   ├── __init__.py
│   └── settings.py          # 配置模块（模块级单例）
├── core/
│   ├── __init__.py
│   ├── database.py          # 数据库模块（模块级单例）
│   └── cache.py             # 缓存模块
├── models/
│   ├── __init__.py
│   └── user.py              # 数据模型（类）
├── repositories/
│   ├── __init__.py
│   └── user_repository.py   # 数据访问层（类 + DI）
├── services/
│   ├── __init__.py
│   └── user_service.py      # 业务逻辑层（类 + DI）
├── container.py             # DI 容器
└── main.py                  # 入口文件

6.1.2、配置层：使用模块级单例

# config/settings.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """应用配置（使用 Pydantic Settings）"""
    
    database_url: str = "postgresql://localhost:5432/mydb"
    redis_url: str = "redis://localhost:6379"
    debug: bool = False
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache()
def get_settings() -> Settings:
    """获取配置单例（使用 lru_cache 实现单例）"""
    return Settings()

settings = get_settings()

为什么用模块级单例？

• 配置信息全局唯一
• 不需要复杂的状态管理
• 简单、直观、Pythonic

6.1.3、基础设施层：模块级单例

# core/database.py
import threading
from typing import Optional
from config.settings import settings

class _Database:
    """数据库连接类（私有）"""
    
    def __init__(self):
        self._connection: Optional[any] = None
        self._lock = threading.Lock()
    
    def connect(self) -> None:
        """连接数据库（线程安全）"""
        with self._lock:
            if self._connection is None:
                print(f"Connecting to {settings.database_url}")
                self._connection = f"Connection to {settings.database_url}"
    
    def query(self, sql: str) -> list:
        """执行查询"""
        if self._connection is None:
            self.connect()
        print(f"Executing: {sql}")
        return [{"id": 1, "name": "Alice", "email": "alice@example.com"}]
    
    def close(self) -> None:
        """关闭连接"""
        with self._lock:
            if self._connection:
                print("Closing database connection")
                self._connection = None

db = _Database()

# core/cache.py
from typing import Any, Optional
import threading
from config.settings import settings

class _RedisCache:
    """Redis 缓存类（私有）"""
    
    def __init__(self):
        self._cache: dict[str, Any] = {}
        self._lock = threading.Lock()
    
    def get(self, key: str) -> Optional[Any]:
        """获取缓存"""
        with self._lock:
            return self._cache.get(key)
    
    def set(self, key: str, value: Any, ttl: int = 3600) -> None:
        """设置缓存"""
        with self._lock:
            self._cache[key] = value
            print(f"Cache set: {key} = {value}")
    
    def delete(self, key: str) -> None:
        """删除缓存"""
        with self._lock:
            self._cache.pop(key, None)

cache = _RedisCache()

为什么用模块级单例？

• 数据库连接、缓存连接通常全局唯一
• 需要线程安全（模块导入自动保证）
• 生命周期贯穿整个应用

6.1.4、数据模型层：使用类

# models/user.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class User:
    """用户模型"""
    id: int
    name: str
    email: str
    age: Optional[int] = None
    
    def is_adult(self) -> bool:
        """判断是否成年"""
        return self.age is not None and self.age >= 18

为什么用类？

• 需要封装数据和行为
• 需要多个实例
• 符合面向对象设计

6.1.5、数据访问层：类 + 依赖注入

# repositories/user_repository.py
from typing import Protocol, Optional, List
from models.user import User

class DatabaseProtocol(Protocol):
    """数据库接口协议"""
    
    def query(self, sql: str) -> list:
        """执行查询"""
        ...

class UserRepository:
    """用户仓储（通过 DI 注入数据库依赖）"""
    
    def __init__(self, database: DatabaseProtocol):
        """构造函数注入数据库依赖"""
        self.db = database
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        """根据 ID 查找用户"""
        result = self.db.query(f"SELECT * FROM users WHERE id = {user_id}")
        if result:
            return User(**result[0])
        return None
    
    def find_all(self) -> List[User]:
        """查找所有用户"""
        result = self.db.query("SELECT * FROM users")
        return [User(**row) for row in result]
    
    def save(self, user: User) -> None:
        """保存用户"""
        self.db.query(f"INSERT INTO users (name, email) VALUES ('{user.name}', '{user.email}')")
    
    def delete(self, user_id: int) -> None:
        """删除用户"""
        self.db.query(f"DELETE FROM users WHERE id = {user_id}")

为什么用依赖注入？

• 易于测试（可以注入 mock 数据库）
• 低耦合（依赖接口而非实现）
• 灵活（可以切换不同的数据库实现）

6.1.6、业务逻辑层：类 + 依赖注入

# services/user_service.py
from typing import Protocol, Optional, List
from models.user import User

class UserRepositoryProtocol(Protocol):
    """用户仓储接口协议"""
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        ...
    
    def find_all(self) -> List[User]:
        ...
    
    def save(self, user: User) -> None:
        ...
    
    def delete(self, user_id: int) -> None:
        ...

class CacheProtocol(Protocol):
    """缓存接口协议"""
    
    def get(self, key: str):
        ...
    
    def set(self, key: str, value, ttl: int = 3600) -> None:
        ...
    
    def delete(self, key: str) -> None:
        ...

class UserService:
    """用户服务（业务逻辑层）"""
    
    def __init__(self, repository: UserRepositoryProtocol, cache: CacheProtocol):
        """构造函数注入依赖"""
        self.repository = repository
        self.cache = cache
    
    def get_user(self, user_id: int) -> Optional[User]:
        """获取用户（带缓存）"""
        cache_key = f"user:{user_id}"
        
        cached = self.cache.get(cache_key)
        if cached:
            print(f"Cache hit: {cache_key}")
            return cached
        
        user = self.repository.find_by_id(user_id)
        if user:
            self.cache.set(cache_key, user)
        
        return user
    
    def create_user(self, name: str, email: str) -> User:
        """创建用户"""
        if "@" not in email:
            raise ValueError("Invalid email format")
        
        user = User(id=0, name=name, email=email)
        self.repository.save(user)
        return user
    
    def delete_user(self, user_id: int) -> None:
        """删除用户（删除缓存）"""
        self.repository.delete(user_id)
        self.cache.delete(f"user:{user_id}")

为什么用依赖注入？

• 业务逻辑需要多个依赖（Repository、Cache）
• 需要灵活切换不同的实现
• 易于单元测试

6.1.7、DI 容器：统一管理依赖

# container.py
from dependency_injector import containers, providers
from config.settings import settings
from core.database import db
from core.cache import cache
from repositories.user_repository import UserRepository
from services.user_service import UserService

class Container(containers.DeclarativeContainer):
    """依赖注入容器"""
    
    config = providers.Configuration()
    
    database = providers.Singleton(lambda: db)
    
    cache_service = providers.Singleton(lambda: cache)
    
    user_repository = providers.Singleton(
        UserRepository,
        database=database,
    )
    
    user_service = providers.Factory(
        UserService,
        repository=user_repository,
        cache=cache_service,
    )

def create_container() -> Container:
    """创建并配置容器"""
    container = Container()
    return container

为什么用 DI 容器？

• 统一管理依赖关系
• 自动解析依赖链
• 配置集中化
• 生命周期管理（单例 vs 瞬态）

6.1.8、入口文件：组装所有组件

# main.py
from container import create_container
from core.database import db

def main():
    """主函数"""
    container = create_container()
    
    db.connect()
    
    try:
        user_service = container.user_service()
        
        print("=== 创建用户 ===")
        user = user_service.create_user("Bob", "bob@example.com")
        print(f"Created: {user}")
        
        print("\n=== 获取用户（首次，无缓存） ===")
        user = user_service.get_user(1)
        print(f"User: {user}")
        
        print("\n=== 获取用户（第二次，有缓存） ===")
        user = user_service.get_user(1)
        print(f"User: {user}")
        
        print("\n=== 删除用户 ===")
        user_service.delete_user(1)
        print("User deleted")
        
    finally:
        db.close()

if __name__ == "__main__":
    main()

输出：

Connecting to postgresql://localhost:5432/mydb
=== 创建用户 ===
Executing: INSERT INTO users (name, email) VALUES ('Bob', 'bob@example.com')
Created: User(id=0, name='Bob', email='bob@example.com', age=None)

=== 获取用户（首次，无缓存） ===
Executing: SELECT * FROM users WHERE id = 1
Cache set: user:1 = User(id=1, name='Alice', email='alice@example.com', age=None)
User: User(id=1, name='Alice', email='alice@example.com', age=None)

=== 获取用户（第二次，有缓存） ===
Cache hit: user:1
User: User(id=1, name='Alice', email='alice@example.com', age=None)

=== 删除用户 ===
Executing: DELETE FROM users WHERE id = 1
User deleted
Closing database connection

6.2、测试：依赖注入的威力

# tests/test_user_service.py
from unittest.mock import Mock
import pytest
from models.user import User
from services.user_service import UserService

def test_get_user_with_cache():
    """测试获取用户（有缓存）"""
    mock_repository = Mock()
    mock_cache = Mock()
    
    mock_cache.get.return_value = User(id=1, name="Alice", email="alice@example.com")
    
    service = UserService(repository=mock_repository, cache=mock_cache)
    
    user = service.get_user(1)
    
    assert user.name == "Alice"
    mock_cache.get.assert_called_once_with("user:1")
    mock_repository.find_by_id.assert_not_called()

def test_get_user_without_cache():
    """测试获取用户（无缓存）"""
    mock_repository = Mock()
    mock_cache = Mock()
    
    mock_cache.get.return_value = None
    mock_repository.find_by_id.return_value = User(id=1, name="Bob", email="bob@example.com")
    
    service = UserService(repository=mock_repository, cache=mock_cache)
    
    user = service.get_user(1)
    
    assert user.name == "Bob"
    mock_cache.get.assert_called_once_with("user:1")
    mock_repository.find_by_id.assert_called_once_with(1)
    mock_cache.set.assert_called_once()

def test_create_user_invalid_email():
    """测试创建用户（无效邮箱）"""
    mock_repository = Mock()
    mock_cache = Mock()
    
    service = UserService(repository=mock_repository, cache=mock_cache)
    
    with pytest.raises(ValueError, match="Invalid email format"):
        service.create_user("Alice", "invalid-email")

测试的优点：

• 完全隔离外部依赖（数据库、缓存）
• 测试速度快（无 IO 操作）
• 可以精确控制测试场景
• 可以验证方法调用行为

6.3、架构决策总结

层级	使用方案	原因
配置层	模块级单例	全局唯一、简单
基础设施层	模块级单例	数据库/缓存连接全局唯一
数据模型层	类	需要多实例、封装数据和行为
数据访问层	类 + DI	需要依赖数据库、易于测试
业务逻辑层	类 + DI	复杂依赖关系、易于测试
依赖管理	DI 容器	统一管理、自动解析

关键原则：

1. 简单优先：能用模块就用模块，不要过度设计
2. 职责分离：配置、基础设施、业务逻辑分层清晰
3. 依赖倒置：高层模块依赖接口（Protocol），而非具体实现
4. 可测试性：通过 DI 让每一层都易于测试
5. 生命周期管理：通过 DI 容器统一管理单例和瞬态对象

七、总结

7.1、核心概念回顾

7.1.1、模块 vs 类

维度	模块	类
本质	天然单例	可以多实例
使用场景	配置、工具函数、全局单例	业务逻辑、数据模型
复杂度	⭐	⭐⭐
推荐度	简单场景首选	复杂场景首选

决策规则：

• 需要多个实例？→ 用类
• 需要复杂状态管理？→ 用类（单例模式）
• 只需要工具函数或简单配置？→ 用模块

7.1.2、单例模式

实现方式	线程安全	延迟初始化	复杂度	推荐场景
模块级单例	✅	❌	⭐	90% 场景首选
装饰器单例	✅	✅	⭐⭐	需要延迟初始化
元类单例	✅	✅	⭐⭐⭐	需要继承单例特性

关键知识点：

1. 模块本身就是单例，Python 解释器保证线程安全
2. 类单例需要手动处理线程安全（双重检查锁定）
3. 优先使用模块级单例，简单且 Pythonic

7.1.3、依赖注入

为什么需要 DI？

• ❌ 测试困难：全局单例难以 mock
• ❌ 高耦合：代码与具体实现绑定
• ❌ 不灵活：无法动态切换实现

DI 的三种方式：

1. 构造函数注入（推荐）：依赖关系清晰，易于测试
2. 属性注入：灵活但容易出错
3. 方法注入：适合临时依赖

DI 容器的价值：

• 统一管理依赖关系
• 自动解析依赖链
• 生命周期管理（单例 vs 瞬态）
• 配置集中化

7.2、架构设计决策指南

7.2.1、按场景选择方案

场景	推荐方案	理由
配置信息	模块级单例	全局唯一、简单
数据库连接	模块级单例	全局唯一、线程安全
缓存服务	模块级单例	全局唯一、线程安全
工具函数	模块级函数	无状态、简单
数据模型	类（dataclass/Pydantic）	多实例、类型安全
业务逻辑	类 + DI	复杂依赖、易于测试
数据访问层	类 + DI	需要依赖数据库、易于测试

7.2.2、按项目规模选择方案

小型项目（< 10 个文件）：

project/
├── config.py          # 模块级配置
├── database.py        # 模块级数据库
├── models.py          # 数据模型
├── service.py         # 业务逻辑（可选 DI）
└── main.py

• 配置、数据库：模块级单例
• 业务逻辑：简单类，手动注入依赖

中型项目（10-50 个文件）：

project/
├── config/
│   └── settings.py    # 模块级配置
├── core/
│   ├── database.py    # 模块级数据库
│   └── cache.py       # 模块级缓存
├── models/
├── repositories/      # 类 + DI
├── services/          # 类 + DI
├── container.py       # DI 容器
└── main.py

• 基础设施：模块级单例
• 业务逻辑：类 + DI
• 依赖管理：DI 容器

大型项目（> 50 个文件）：

project/
├── config/
├── core/
├── domain/            # 领域模型
├── application/       # 应用服务
├── infrastructure/    # 基础设施
├── container.py       # DI 容器
└── main.py

• 采用 DDD（领域驱动设计）
• 完整的 DI 容器
• 严格的分层架构

7.3、最佳实践清单

7.3.1、设计原则

• ✅ KISS 原则：能用模块就用模块，不要过度设计
• ✅ 单一职责：每个模块/类只做一件事
• ✅ 依赖倒置：依赖接口（Protocol），而非具体实现
• ✅ 开闭原则：对扩展开放，对修改封闭
• ✅ 测试驱动：设计时考虑可测试性

7.3.2、代码规范

# ✅ 推荐：模块级单例
# config.py
class _Config:
    def __init__(self):
        self.database_url = "postgresql://localhost/mydb"

config = _Config()

# ✅ 推荐：类 + 构造函数 DI
class UserService:
    def __init__(self, repository: UserRepositoryProtocol):
        self.repository = repository

# ✅ 推荐：使用 Protocol 定义接口
from typing import Protocol

class DatabaseProtocol(Protocol):
    def query(self, sql: str) -> list:
        ...

# ❌ 避免：全局导入依赖
from database import db  # 不好

class UserService:
    def get_user(self, user_id: int):
        return db.query(...)  # 难以测试

# ✅ 改进：依赖注入
class UserService:
    def __init__(self, database: DatabaseProtocol):
        self.db = database
    
    def get_user(self, user_id: int):
        return self.db.query(...)  # 易于测试

7.3.3、测试策略

# ✅ 单元测试：mock 所有依赖
def test_user_service():
    mock_repository = Mock()
    mock_cache = Mock()
    
    service = UserService(
        repository=mock_repository,
        cache=mock_cache
    )
    
    # 测试业务逻辑

# ✅ 集成测试：使用真实依赖
def test_user_service_integration():
    from core.database import db
    from core.cache import cache
    
    repository = UserRepository(database=db)
    service = UserService(
        repository=repository,
        cache=cache
    )
    
    # 测试完整流程

7.4、常见误区

误区 1：过度使用类

# ❌ 不推荐：简单配置也用类
class Config:
    DATABASE_URL = "postgresql://localhost/mydb"
    REDIS_URL = "redis://localhost:6379"

config = Config()

# ✅ 推荐：直接用模块
DATABASE_URL = "postgresql://localhost/mydb"
REDIS_URL = "redis://localhost:6379"

误区 2：不使用依赖注入

# ❌ 不推荐：全局导入
from database import db

class UserService:
    def get_user(self, user_id: int):
        return db.query(...)  # 难以测试

# ✅ 推荐：依赖注入
class UserService:
    def __init__(self, database: DatabaseProtocol):
        self.db = database
    
    def get_user(self, user_id: int):
        return self.db.query(...)  # 易于测试

误区 3：忽略线程安全

# ❌ 不推荐：非线程安全的单例
class Singleton:
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)  # 可能创建多个实例
        return cls._instance

# ✅ 推荐：线程安全的单例
import threading

class Singleton:
    _instance = None
    _lock = threading.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:  # 双重检查
                    cls._instance = super().__new__(cls)
        return cls._instance

# ✅ 更推荐：模块级单例（天然线程安全）
class _Singleton:
    pass

singleton = _Singleton()

误区 4：过度依赖 DI 容器

# ❌ 不推荐：简单项目也用复杂的 DI 容器
from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()
    # 大量配置...

# ✅ 推荐：简单项目手动注入
database = Database()
repository = UserRepository(database)
service = UserService(repository)

7.5、进阶学习路径

1. 类型系统：

   • 深入学习 typing 模块
   • 使用 mypy 进行静态类型检查
   • 掌握 Protocol 和 TypeVar

2. 异步编程：

   • 学习 asyncio 和 async/await
   • 异步依赖注入（FastAPI）
   • 异步上下文管理器

3. 设计模式：

   • 工厂模式、策略模式、观察者模式
   • Python 特有的设计模式
   • 函数式编程模式

4. 框架深入：

   • FastAPI 的依赖注入机制
   • Django 的 ORM 和中间件
   • Flask 的应用上下文

5. 架构模式：

   • 领域驱动设计（DDD）
   • CQRS（命令查询职责分离）
   • 事件驱动架构

7.6、总结一句话

简单的问题用简单的方案（模块），复杂的问题用合适的工具（类 + DI），永远保持 KISS 原则。

祝你变得更强!