跳转至

DjangoContainer

symphra_container.integrations.DjangoContainer

Django 应用的容器包装器.

提供与 Django 请求生命周期集成的依赖注入功能。

属性：

名称 类型 描述
_container Container | None

全局容器实例
示例

在 settings.py 中

from django.conf import settings settings.CONTAINER = Container()

在视图中

user_service = DjangoContainer.resolve(UserService)

Source code in src/symphra_container/integrations/django.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class DjangoContainer:
    """Django 应用的容器包装器.

    提供与 Django 请求生命周期集成的依赖注入功能。

    Attributes:
        _container: 全局容器实例

    示例:
        >>> # 在 settings.py 中
        >>> from django.conf import settings
        >>> settings.CONTAINER = Container()
        >>> # 在视图中
        >>> user_service = DjangoContainer.resolve(UserService)
    """

    _container: Container | None = None

    @classmethod
    def setup(cls, container: Container) -> None:
        """设置全局容器实例.

        通常在 settings.py 或应用启动时调用。

        Args:
            container: 容器实例

        Raises:
            ImportError: 如果未安装 Django

        示例:
            >>> from django.conf import settings
            >>> container = Container()
            >>> DjangoContainer.setup(container)
        """
        try:
            import django  # noqa: F401
        except ImportError as e:
            raise ImportError("Django is not installed. Install it with: pip install symphra-container[django]") from e

        cls._container = container

    @classmethod
    def get_container(cls) -> Container:
        """获取全局容器实例.

        Returns:
            Container: 容器实例

        Raises:
            RuntimeError: 如果容器未初始化

        示例:
            >>> container = DjangoContainer.get_container()
        """
        if cls._container is None:
            # 尝试从 Django settings 获取
            try:
                from django.conf import settings

                if hasattr(settings, "CONTAINER"):
                    cls._container = settings.CONTAINER
            except Exception:
                pass

        if cls._container is None:
            raise RuntimeError("Container not initialized. Call DjangoContainer.setup() or set settings.CONTAINER")

        return cls._container

    @classmethod
    def resolve(cls, service_type: type[T]) -> T:
        """解析服务实例.

        在请求上下文中使用作用域容器(如果有)，否则使用根容器。

        Args:
            service_type: 要解析的服务类型

        Returns:
            T: 服务实例

        Raises:
            RuntimeError: 如果容器未初始化
            ServiceNotFoundError: 如果服务未注册

        示例:
            >>> user_service = DjangoContainer.resolve(UserService)
        """
        # 尝试从当前线程获取请求对象
        try:
            import threading

            local = threading.local()
            request: HttpRequest | None = getattr(local, "request", None)
            if request and hasattr(request, "container_scope"):
                return request.container_scope.resolve(service_type)
        except Exception:
            pass

        # 使用根容器
        container = cls.get_container()
        return container.resolve(service_type)

    @classmethod
    def inject(cls, func: F) -> F:
        """装饰器: 自动注入函数参数.

        分析函数签名(跳过 request 参数)，根据类型注解自动注入服务。

        Args:
            func: 要装饰的函数(Django 视图函数)

        Returns:
            F: 装饰后的函数

        Raises:
            RuntimeError: 如果服务解析失败

        示例:
            >>> @DjangoContainer.inject
            ... def view(request, user_service: UserService):
            ...     return JsonResponse(user_service.get_all())
        """
        sig = inspect.signature(func)

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            # 分析需要注入的参数
            bound_args = sig.bind_partial(*args, **kwargs)
            bound_args.apply_defaults()

            for param_name, param in sig.parameters.items():
                # 跳过 request 参数和已提供的参数
                if param_name == "request" or param_name in bound_args.arguments:
                    continue

                # 检查是否有类型注解
                if param.annotation == inspect.Parameter.empty:
                    continue

                # 尝试解析服务
                try:
                    service = cls.resolve(param.annotation)
                    bound_args.arguments[param_name] = service
                except Exception:
                    # 无法解析，可能不是容器管理的服务
                    continue

            return func(*bound_args.args, **bound_args.kwargs)

        return cast(F, wrapper)

get_container() classmethod

获取全局容器实例.

返回：

名称 类型 描述
Container Container

容器实例

引发：

类型 描述
RuntimeError

如果容器未初始化
示例

container = DjangoContainer.get_container()

源代码位于： src/symphra_container/integrations/django.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def get_container(cls) -> Container:
    """获取全局容器实例.

    Returns:
        Container: 容器实例

    Raises:
        RuntimeError: 如果容器未初始化

    示例:
        >>> container = DjangoContainer.get_container()
    """
    if cls._container is None:
        # 尝试从 Django settings 获取
        try:
            from django.conf import settings

            if hasattr(settings, "CONTAINER"):
                cls._container = settings.CONTAINER
        except Exception:
            pass

    if cls._container is None:
        raise RuntimeError("Container not initialized. Call DjangoContainer.setup() or set settings.CONTAINER")

    return cls._container

inject(func) classmethod

装饰器: 自动注入函数参数.

分析函数签名(跳过 request 参数)，根据类型注解自动注入服务。

参数：

名称 类型 描述 默认
func F

要装饰的函数(Django 视图函数)

 必需

返回：

名称 类型 描述
F F

装饰后的函数

引发：

类型 描述
RuntimeError

如果服务解析失败
示例

@DjangoContainer.inject ... def view(request, user_service: UserService): ... return JsonResponse(user_service.get_all())

源代码位于： src/symphra_container/integrations/django.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
@classmethod
def inject(cls, func: F) -> F:
    """装饰器: 自动注入函数参数.

    分析函数签名(跳过 request 参数)，根据类型注解自动注入服务。

    Args:
        func: 要装饰的函数(Django 视图函数)

    Returns:
        F: 装饰后的函数

    Raises:
        RuntimeError: 如果服务解析失败

    示例:
        >>> @DjangoContainer.inject
        ... def view(request, user_service: UserService):
        ...     return JsonResponse(user_service.get_all())
    """
    sig = inspect.signature(func)

    @functools.wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        # 分析需要注入的参数
        bound_args = sig.bind_partial(*args, **kwargs)
        bound_args.apply_defaults()

        for param_name, param in sig.parameters.items():
            # 跳过 request 参数和已提供的参数
            if param_name == "request" or param_name in bound_args.arguments:
                continue

            # 检查是否有类型注解
            if param.annotation == inspect.Parameter.empty:
                continue

            # 尝试解析服务
            try:
                service = cls.resolve(param.annotation)
                bound_args.arguments[param_name] = service
            except Exception:
                # 无法解析，可能不是容器管理的服务
                continue

        return func(*bound_args.args, **bound_args.kwargs)

    return cast(F, wrapper)

resolve(service_type) classmethod

解析服务实例.

在请求上下文中使用作用域容器(如果有)，否则使用根容器。

参数：

名称 类型 描述 默认
service_type type[T]

要解析的服务类型

 必需

返回：

名称 类型 描述
T T

服务实例

引发：

类型 描述
RuntimeError

如果容器未初始化
ServiceNotFoundError

如果服务未注册
示例

user_service = DjangoContainer.resolve(UserService)

源代码位于： src/symphra_container/integrations/django.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@classmethod
def resolve(cls, service_type: type[T]) -> T:
    """解析服务实例.

    在请求上下文中使用作用域容器(如果有)，否则使用根容器。

    Args:
        service_type: 要解析的服务类型

    Returns:
        T: 服务实例

    Raises:
        RuntimeError: 如果容器未初始化
        ServiceNotFoundError: 如果服务未注册

    示例:
        >>> user_service = DjangoContainer.resolve(UserService)
    """
    # 尝试从当前线程获取请求对象
    try:
        import threading

        local = threading.local()
        request: HttpRequest | None = getattr(local, "request", None)
        if request and hasattr(request, "container_scope"):
            return request.container_scope.resolve(service_type)
    except Exception:
        pass

    # 使用根容器
    container = cls.get_container()
    return container.resolve(service_type)

setup(container) classmethod

设置全局容器实例.

通常在 settings.py 或应用启动时调用。

参数：

名称 类型 描述 默认
container Container

容器实例

 必需

引发：

类型 描述
ImportError

如果未安装 Django
示例

from django.conf import settings container = Container() DjangoContainer.setup(container)

源代码位于： src/symphra_container/integrations/django.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def setup(cls, container: Container) -> None:
    """设置全局容器实例.

    通常在 settings.py 或应用启动时调用。

    Args:
        container: 容器实例

    Raises:
        ImportError: 如果未安装 Django

    示例:
        >>> from django.conf import settings
        >>> container = Container()
        >>> DjangoContainer.setup(container)
    """
    try:
        import django  # noqa: F401
    except ImportError as e:
        raise ImportError("Django is not installed. Install it with: pip install symphra-container[django]") from e

    cls._container = container