Ellar DI module refactor for easy code review and readability

eadwinCode · eadwinCode · commit 7ba4f99edd8c · 2022-12-23T17:44:52.000+01:00
diff --git a/ellar/constants.py b/ellar/constants.py
@@ -68,6 +68,7 @@ def __new__(mcls, name, bases, namespace):
 CONTROLLER_WATERMARK = "CONTROLLER_WATERMARK"
 
 MULTI_RESOLVER_KEY = "MULTI_RESOLVER_KEY"
+MULTI_RESOLVER_FORM_GROUPED_KEY = "MULTI_RESOLVER_FORM_GROUPED_KEY"
 ROUTE_OPENAPI_PARAMETERS = "ROUTE_OPENAPI_PARAMETERS"
 
 OPERATION_ENDPOINT_KEY = "OPERATION_ENDPOINT"
diff --git a/ellar/di/injector/__init__.py b/ellar/di/injector/__init__.py
@@ -0,0 +1,5 @@
+from .container import Container
+from .ellar_injector import EllarInjector
+from .request_provider import RequestServiceProvider
+
+__all__ = ["Container", "EllarInjector", "RequestServiceProvider"]
diff --git a/ellar/di/injector/container.py b/ellar/di/injector/container.py
@@ -1,83 +1,36 @@
 import typing as t
-from collections import OrderedDict, defaultdict
 from inspect import isabstract
 
-from injector import (
-    Binder as InjectorBinder,
-    Binding,
-    Injector,
-    Module as InjectorModule,
-)
+from injector import Binder as InjectorBinder, Binding, Module as InjectorModule
 
-from ellar.compatible import asynccontextmanager
-from ellar.constants import MODULE_REF_TYPES
 from ellar.helper import get_name
-from ellar.logger import logger as log
 from ellar.types import T
 
-from .providers import InstanceProvider, Provider
-from .scopes import (
+from ..providers import Provider
+from ..scopes import (
     DIScope,
     RequestScope,
     ScopeDecorator,
     SingletonScope,
     TransientScope,
 )
-from .service_config import get_scope
+from ..service_config import get_scope
 
 if t.TYPE_CHECKING:  # pragma: no cover
-    from ellar.core.modules import ModuleBase, ModuleRefBase, ModuleTemplateRef
-
-
-class RequestServiceProvider(InjectorBinder):
-    __slots__ = ("_bindings", "_log_prefix", "_context")
+    from ellar.core.modules import ModuleBase
 
-    def __init__(self, container: "Container", auto_bind: bool = False) -> None:
-        super(RequestServiceProvider, self).__init__(
-            injector=container.injector, parent=container, auto_bind=auto_bind
-        )
-        self._context: t.Dict = {}
-        self._log_prefix = container.injector._log_prefix
-
-    def get(self, interface: t.Type[T]) -> T:
-        binding, binder = self.get_binding(interface)
-        scope = binding.scope
-        if isinstance(scope, ScopeDecorator):
-            scope = scope.scope
-        # Fetch the corresponding Scope instance from the Binder.
-        scope_binding, _ = binder.get_binding(scope)
-        scope_instance = t.cast(DIScope, scope_binding.provider.get(self))
-
-        log.debug(
-            "%EllarInjector.get(%r, scope=%r) using %r",
-            self._log_prefix,
-            interface,
-            scope,
-            binding.provider,
-        )
-        result = scope_instance.get(
-            interface, binding.provider, context=self._context
-        ).get(self.injector)
-        log.debug("%s -> %r", self._log_prefix, result)
-        return t.cast(T, result)
-
-    def update_context(self, interface: t.Type[T], value: T) -> None:
-        assert not isinstance(value, type), f"value must be an object of {interface}"
-        _context = {
-            interface: Binding(interface, InstanceProvider(value), RequestScope)
-        }
-        if isinstance(value, Provider):
-            _context = {interface: Binding(interface, value, RequestScope)}
-        self._bindings.update(_context)
-
-    def dispose(self) -> None:
-        del self._context
-        del self.parent
-        del self.injector
+    from .ellar_injector import EllarInjector
 
 
 class Container(InjectorBinder):
-    __slots__ = ("injector", "_auto_bind", "_bindings", "parent")
+    __slots__ = (
+        "injector",
+        "_auto_bind",
+        "_bindings",
+        "parent",
+        "_aliases",
+        "_exact_aliases",
+    )
 
     injector: "EllarInjector"
 
@@ -132,6 +85,12 @@ def register_singleton(
         base_type: t.Type[T],
         concrete_type: t.Union[t.Type[T], T, Provider] = None,
     ) -> None:
+        """
+
+        :param base_type:
+        :param concrete_type:
+        :return:
+        """
         if not concrete_type:
             self.register_exact_singleton(base_type)
         self.register(base_type, concrete_type, scope=SingletonScope)
@@ -141,6 +100,12 @@ def register_transient(
         base_type: t.Type,
         concrete_type: t.Union[t.Type, Provider] = None,
     ) -> None:
+        """
+
+        :param base_type:
+        :param concrete_type:
+        :return:
+        """
         if not concrete_type:
             self.register_exact_transient(base_type)
         self.register(base_type, concrete_type, scope=TransientScope)
@@ -150,19 +115,40 @@ def register_scoped(
         base_type: t.Type,
         concrete_type: t.Union[t.Type, Provider] = None,
     ) -> None:
+        """
+
+        :param base_type:
+        :param concrete_type:
+        :return:
+        """
         if not concrete_type:
             self.register_exact_scoped(base_type)
         self.register(base_type, concrete_type, scope=RequestScope)
 
     def register_exact_singleton(self, concrete_type: t.Type) -> None:
+        """
+
+        :param concrete_type:
+        :return:
+        """
         assert not isabstract(concrete_type)
         self.register(base_type=concrete_type, scope=SingletonScope)
 
     def register_exact_transient(self, concrete_type: t.Type) -> None:
+        """
+
+        :param concrete_type:
+        :return:
+        """
         assert not isabstract(concrete_type)
         self.register(base_type=concrete_type, scope=TransientScope)
 
     def register_exact_scoped(self, concrete_type: t.Type) -> None:
+        """
+
+        :param concrete_type:
+        :return:
+        """
         assert not isabstract(concrete_type)
         self.register(base_type=concrete_type, scope=RequestScope)
 
@@ -215,74 +201,3 @@ def register_services(self, container):
 
         instance(self)
         return instance
-
-
-class EllarInjector(Injector):
-    __slots__ = ("_stack", "parent", "app", "container", "_modules")
-
-    def __init__(
-        self,
-        auto_bind: bool = True,
-        parent: "Injector" = None,
-    ) -> None:
-        self._stack = ()
-
-        self.parent = parent
-        # Binder
-        self.container = Container(
-            self,
-            auto_bind=auto_bind,
-            parent=parent.binder if parent is not None else None,
-        )
-        # Bind some useful types
-        self.container.register_instance(self, EllarInjector)
-        self.container.register_instance(self.binder)
-        self._modules: t.DefaultDict = defaultdict(OrderedDict)
-        self._modules[MODULE_REF_TYPES.TEMPLATE] = OrderedDict()
-        self._modules[MODULE_REF_TYPES.PLAIN] = OrderedDict()
-
-    def get_modules(
-        self,
-    ) -> t.Dict[t.Type["ModuleBase"], "ModuleRefBase"]:
-        modules = dict(
-            self._modules[MODULE_REF_TYPES.TEMPLATE],
-        )
-        modules.update(self._modules[MODULE_REF_TYPES.PLAIN])
-        return modules
-
-    def get_module(self, module: t.Type) -> t.Optional["ModuleRefBase"]:
-        result: t.Optional["ModuleRefBase"] = None
-        if module in self._modules[MODULE_REF_TYPES.TEMPLATE]:
-            result = self._modules[MODULE_REF_TYPES.TEMPLATE][module]
-            return result
-
-        if module in self._modules[MODULE_REF_TYPES.PLAIN]:
-            result = self._modules[MODULE_REF_TYPES.PLAIN][module]
-            return result
-        return result
-
-    def get_templating_modules(
-        self,
-    ) -> t.Dict[t.Type["ModuleBase"], "ModuleTemplateRef"]:
-        return self._modules.get(MODULE_REF_TYPES.TEMPLATE, {})
-
-    def add_module(self, module_ref: "ModuleRefBase") -> None:
-        self._modules[module_ref.ref_type].update({module_ref.module: module_ref})
-
-    @property  # type: ignore
-    def binder(self) -> Container:  # type: ignore
-        return self.container
-
-    @binder.setter
-    def binder(self, value: t.Any) -> None:
-        """Nothing happens"""
-
-    @asynccontextmanager
-    async def create_request_service_provider(
-        self,
-    ) -> t.AsyncGenerator[RequestServiceProvider, None]:
-        request_provider = RequestServiceProvider(self.container)
-        try:
-            yield request_provider
-        finally:
-            request_provider.dispose()
diff --git a/ellar/di/injector/ellar_injector.py b/ellar/di/injector/ellar_injector.py
@@ -0,0 +1,85 @@
+import typing as t
+from collections import OrderedDict, defaultdict
+
+from injector import Injector
+
+from ellar.compatible import asynccontextmanager
+from ellar.constants import MODULE_REF_TYPES
+
+from .container import Container
+from .request_provider import RequestServiceProvider
+
+if t.TYPE_CHECKING:  # pragma: no cover
+    from ellar.core.modules import ModuleBase, ModuleRefBase, ModuleTemplateRef
+
+
+class EllarInjector(Injector):
+    __slots__ = ("_stack", "parent", "container", "_modules")
+
+    def __init__(
+        self,
+        auto_bind: bool = True,
+        parent: "Injector" = None,
+    ) -> None:
+        self._stack = ()
+
+        self.parent = parent
+        # Binder
+        self.container = Container(
+            self,
+            auto_bind=auto_bind,
+            parent=parent.binder if parent is not None else None,
+        )
+
+        # Bind some useful types
+        self.container.register_instance(self, EllarInjector)
+        self.container.register_instance(self.binder)
+        self._modules: t.DefaultDict = defaultdict(OrderedDict)
+        self._modules[MODULE_REF_TYPES.TEMPLATE] = OrderedDict()
+        self._modules[MODULE_REF_TYPES.PLAIN] = OrderedDict()
+
+    @property  # type: ignore
+    def binder(self) -> Container:  # type: ignore
+        return self.container
+
+    @binder.setter
+    def binder(self, value: t.Any) -> None:
+        """Nothing happens"""
+
+    def get_modules(
+        self,
+    ) -> t.Dict[t.Type["ModuleBase"], "ModuleRefBase"]:
+        modules = dict(
+            self._modules[MODULE_REF_TYPES.TEMPLATE],
+        )
+        modules.update(self._modules[MODULE_REF_TYPES.PLAIN])
+        return modules
+
+    def get_module(self, module: t.Type) -> t.Optional["ModuleRefBase"]:
+        result: t.Optional["ModuleRefBase"] = None
+        if module in self._modules[MODULE_REF_TYPES.TEMPLATE]:
+            result = self._modules[MODULE_REF_TYPES.TEMPLATE][module]
+            return result
+
+        if module in self._modules[MODULE_REF_TYPES.PLAIN]:
+            result = self._modules[MODULE_REF_TYPES.PLAIN][module]
+            return result
+        return result
+
+    def get_templating_modules(
+        self,
+    ) -> t.Dict[t.Type["ModuleBase"], "ModuleTemplateRef"]:
+        return self._modules.get(MODULE_REF_TYPES.TEMPLATE, {})
+
+    def add_module(self, module_ref: "ModuleRefBase") -> None:
+        self._modules[module_ref.ref_type].update({module_ref.module: module_ref})
+
+    @asynccontextmanager
+    async def create_request_service_provider(
+        self,
+    ) -> t.AsyncGenerator[RequestServiceProvider, None]:
+        request_provider = RequestServiceProvider(self.container)
+        try:
+            yield request_provider
+        finally:
+            request_provider.dispose()
diff --git a/ellar/di/injector/request_provider.py b/ellar/di/injector/request_provider.py
@@ -0,0 +1,59 @@
+import typing as t
+
+from injector import Binder as InjectorBinder, Binding
+
+from ellar.logger import logger as log
+from ellar.types import T
+
+from ..providers import InstanceProvider, Provider
+from ..scopes import DIScope, RequestScope, ScopeDecorator
+
+if t.TYPE_CHECKING:  # pragma: no cover
+    from .container import Container
+
+
+class RequestServiceProvider(InjectorBinder):
+    __slots__ = ("_bindings", "_log_prefix", "_context")
+
+    def __init__(self, container: "Container", auto_bind: bool = False) -> None:
+        super(RequestServiceProvider, self).__init__(
+            injector=container.injector, parent=container, auto_bind=auto_bind
+        )
+        self._context: t.Dict = {}
+        self._log_prefix = container.injector._log_prefix
+
+    def get(self, interface: t.Type[T]) -> T:
+        binding, binder = self.get_binding(interface)
+        scope = binding.scope
+        if isinstance(scope, ScopeDecorator):
+            scope = scope.scope
+        # Fetch the corresponding Scope instance from the Binder.
+        scope_binding, _ = binder.get_binding(scope)
+        scope_instance = t.cast(DIScope, scope_binding.provider.get(self))
+
+        log.debug(
+            "%EllarInjector.get(%r, scope=%r) using %r",
+            self._log_prefix,
+            interface,
+            scope,
+            binding.provider,
+        )
+        result = scope_instance.get(
+            interface, binding.provider, context=self._context
+        ).get(self.injector)
+        log.debug("%s -> %r", self._log_prefix, result)
+        return t.cast(T, result)
+
+    def update_context(self, interface: t.Type[T], value: T) -> None:
+        assert not isinstance(value, type), f"value must be an object of {interface}"
+        _context = {
+            interface: Binding(interface, InstanceProvider(value), RequestScope)
+        }
+        if isinstance(value, Provider):
+            _context = {interface: Binding(interface, value, RequestScope)}
+        self._bindings.update(_context)
+
+    def dispose(self) -> None:
+        del self._context
+        del self.parent
+        del self.injector
