跳转至

DI Container

controller.shared.config.di_container

依赖注入容器 - Shared Config 简洁的DI容器实现,只支持单例和瞬时服务

ServiceLifetime

服务生命周期枚举。

属性:

名称 类型 描述
SINGLETON

单例模式,整个容器生命周期内只创建一个实例。

TRANSIENT

瞬时模式,每次请求都创建新的实例。

ServiceDescriptor

ServiceDescriptor(service_type: Type, implementation_type: Type = None, lifetime: str = ServiceLifetime.SINGLETON)

服务描述符。

用于存储服务的注册信息。

属性:

名称 类型 描述
service_type Type

服务类型(接口或基类)。

implementation_type Type

实现类型(具体类)。

lifetime str

服务生命周期。

初始化服务描述符。

参数:

名称 类型 描述 默认
service_type Type

服务类型。

必需
implementation_type Type

实现类型. Defaults to None.

None
lifetime str

生命周期. Defaults to ServiceLifetime.SINGLETON.

SINGLETON
源代码位于: src/controller/controller/shared/config/di_container.py
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(self, service_type: Type, implementation_type: Type = None, 
             lifetime: str = ServiceLifetime.SINGLETON):
    """初始化服务描述符。

    Args:
        service_type (Type): 服务类型。
        implementation_type (Type, optional): 实现类型. Defaults to None.
        lifetime (str, optional): 生命周期. Defaults to ServiceLifetime.SINGLETON.
    """
    self.service_type = service_type
    self.implementation_type = implementation_type or service_type
    self.lifetime = lifetime

DIContainer

DIContainer()

依赖注入容器。

简洁的 DI 容器实现,只支持单例和瞬时服务。

属性:

名称 类型 描述
_services Dict[Type, ServiceDescriptor]

已注册的服务描述符字典。

_instances Dict[Type, Any]

已创建的单例实例字典。

初始化容器。

源代码位于: src/controller/controller/shared/config/di_container.py
55
56
57
58
def __init__(self):
    """初始化容器。"""
    self._services: Dict[Type, ServiceDescriptor] = {}
    self._instances: Dict[Type, Any] = {}

register_singleton

register_singleton(service_type: Type, implementation_type: Type = None) -> DIContainer

注册单例服务。

参数:

名称 类型 描述 默认
service_type Type

服务类型。

必需
implementation_type Type

实现类型. Defaults to None.

None

返回:

名称 类型 描述
DIContainer DIContainer

容器实例本身(支持链式调用)。

源代码位于: src/controller/controller/shared/config/di_container.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def register_singleton(self, service_type: Type, implementation_type: Type = None) -> 'DIContainer':
    """注册单例服务。

    Args:
        service_type (Type): 服务类型。
        implementation_type (Type, optional): 实现类型. Defaults to None.

    Returns:
        DIContainer: 容器实例本身(支持链式调用)。
    """
    self._services[service_type] = ServiceDescriptor(
        service_type, implementation_type, lifetime=ServiceLifetime.SINGLETON
    )
    return self

register_transient

register_transient(service_type: Type, implementation_type: Type = None) -> DIContainer

注册瞬时服务。

参数:

名称 类型 描述 默认
service_type Type

服务类型。

必需
implementation_type Type

实现类型. Defaults to None.

None

返回:

名称 类型 描述
DIContainer DIContainer

容器实例本身(支持链式调用)。

源代码位于: src/controller/controller/shared/config/di_container.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def register_transient(self, service_type: Type, implementation_type: Type = None) -> 'DIContainer':
    """注册瞬时服务。

    Args:
        service_type (Type): 服务类型。
        implementation_type (Type, optional): 实现类型. Defaults to None.

    Returns:
        DIContainer: 容器实例本身(支持链式调用)。
    """
    self._services[service_type] = ServiceDescriptor(
        service_type, implementation_type, lifetime=ServiceLifetime.TRANSIENT
    )
    return self

get_service

get_service(service_type: Type) -> Any

获取服务实例。

参数:

名称 类型 描述 默认
service_type Type

服务类型。

必需

返回:

名称 类型 描述
Any Any

服务实例。

引发:

类型 描述
ValueError

如果服务未注册。

源代码位于: src/controller/controller/shared/config/di_container.py
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def get_service(self, service_type: Type) -> Any:
    """获取服务实例。

    Args:
        service_type (Type): 服务类型。

    Returns:
        Any: 服务实例。

    Raises:
        ValueError: 如果服务未注册。
    """
    if service_type not in self._services:
        raise ValueError(f"服务 {service_type.__name__} 未注册")

    descriptor = self._services[service_type]

    # 如果是单例且已创建,直接返回
    if descriptor.lifetime == ServiceLifetime.SINGLETON and service_type in self._instances:
        return self._instances[service_type]

    # 创建新实例
    instance = self._create_with_constructor(descriptor.implementation_type)

    # 如果是单例,缓存实例
    if descriptor.lifetime == ServiceLifetime.SINGLETON:
        self._instances[service_type] = instance

    return instance

is_registered

is_registered(service_type: Type) -> bool

检查服务是否已注册。

参数:

名称 类型 描述 默认
service_type Type

服务类型。

必需

返回:

名称 类型 描述
bool bool

是否已注册。

源代码位于: src/controller/controller/shared/config/di_container.py
174
175
176
177
178
179
180
181
182
183
def is_registered(self, service_type: Type) -> bool:
    """检查服务是否已注册。

    Args:
        service_type (Type): 服务类型。

    Returns:
        bool: 是否已注册。
    """
    return service_type in self._services

clear

clear() -> None

清空容器。

清理所有注册的服务和已创建的单例实例。

源代码位于: src/controller/controller/shared/config/di_container.py
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def clear(self) -> None:
    """清空容器。

    清理所有注册的服务和已创建的单例实例。
    """
    # 清理单例实例
    for instance in self._instances.values():
        if hasattr(instance, 'cleanup'):
            try:
                instance.cleanup()
            except Exception:
                pass  # 忽略清理错误

    self._services.clear()
    self._instances.clear()

get_container

get_container() -> DIContainer

获取全局DI容器。

如果容器不存在,则创建一个新实例。

返回:

名称 类型 描述
DIContainer DIContainer

全局容器实例。

源代码位于: src/controller/controller/shared/config/di_container.py
206
207
208
209
210
211
212
213
214
215
216
217
def get_container() -> DIContainer:
    """获取全局DI容器。

    如果容器不存在,则创建一个新实例。

    Returns:
        DIContainer: 全局容器实例。
    """
    global _container
    if _container is None:
        _container = DIContainer()
    return _container

resolve

resolve(service_type: Type) -> Any

便捷方法:解析服务。

从全局容器中获取指定类型的服务实例。

参数:

名称 类型 描述 默认
service_type Type

服务类型。

必需

返回:

名称 类型 描述
Any Any

服务实例。

源代码位于: src/controller/controller/shared/config/di_container.py
220
221
222
223
224
225
226
227
228
229
230
231
def resolve(service_type: Type) -> Any:
    """便捷方法:解析服务。

    从全局容器中获取指定类型的服务实例。

    Args:
        service_type (Type): 服务类型。

    Returns:
        Any: 服务实例。
    """
    return get_container().get_service(service_type)