Add implementation and typing stubs

This commit is contained in:
Roman Mogylatov 2022-01-08 19:16:16 -05:00
parent cfadd8c3fa
commit eb893a756c
5 changed files with 12836 additions and 9588 deletions

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@ -38,6 +38,12 @@ cdef class Delegate(Provider):
cpdef object _provide(self, tuple args, dict kwargs)
cdef class Aggregate(Provider):
cdef dict __providers
cdef Provider __get_provider(self, object provider_name)
cdef class Dependency(Provider):
cdef object __instance_of
cdef object __default

View File

@ -104,6 +104,21 @@ class Delegate(Provider[Provider]):
def set_provides(self, provides: Optional[Provider]) -> Delegate: ...
class Aggregate(Provider[T]):
def __init__(self, provider_dict: Optional[_Dict[Any, Provider[T]]] = None, **provider_kwargs: Provider[T]): ...
def __getattr__(self, provider_name: Any) -> Provider[T]: ...
@overload
def __call__(self, provider_name: Optional[Any] = None, *args: Injection, **kwargs: Injection) -> T: ...
@overload
def __call__(self, provider_name: Optional[Any] = None, *args: Injection, **kwargs: Injection) -> Awaitable[T]: ...
def async_(self, provider_name: Optional[Any] = None, *args: Injection, **kwargs: Injection) -> Awaitable[T]: ...
@property
def providers(self) -> _Dict[Any, Provider[T]]: ...
def set_providers(self, provider_dict: Optional[_Dict[Any, Provider[T]]] = None, **provider_kwargs: Provider[T]) -> Aggregate[T]: ...
class Dependency(Provider[T]):
def __init__(self, instance_of: Type[T] = object, default: Optional[Union[Provider, Any]] = None) -> None: ...
def __getattr__(self, name: str) -> Any: ...

View File

@ -632,6 +632,115 @@ cdef class Delegate(Provider):
return self.__provides
cdef class Aggregate(Provider):
"""Providers aggregate.
:py:class:`Aggregate` is a delegated provider, meaning that it is
injected "as is".
All aggregated providers can be retrieved as a read-only
dictionary :py:attr:`Aggregate.providers` or as an attribute of
:py:class:`Aggregate`, e.g. ``aggregate.provider``.
"""
__IS_DELEGATED__ = True
def __init__(self, provider_dict=None, **provider_kwargs):
"""Initialize provider."""
self.__providers = {}
self.set_providers(provider_dict, **provider_kwargs)
super().__init__()
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
copied = _memorized_duplicate(self, memo)
copied.set_providers(deepcopy(self.providers, memo))
self._copy_overridings(copied, memo)
return copied
def __getattr__(self, factory_name):
"""Return aggregated provider."""
return self.__get_provider(factory_name)
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.providers)
@property
def providers(self):
"""Return dictionary of providers, read-only.
Alias for ``.factories`` attribute.
"""
return dict(self.__providers)
def set_providers(self, provider_dict=None, **provider_kwargs):
"""Set providers.
Alias for ``.set_factories()`` method.
"""
providers = {}
providers.update(provider_kwargs)
if provider_dict:
providers.update(provider_dict)
for provider in providers.values():
if not is_provider(provider):
raise Error(
'{0} can aggregate only instances of {1}, given - {2}'.format(
self.__class__,
Provider,
provider,
),
)
self.__providers = providers
return self
def override(self, _):
"""Override provider with another provider.
:raise: :py:exc:`dependency_injector.errors.Error`
:return: Overriding context.
:rtype: :py:class:`OverridingContext`
"""
raise Error('{0} providers could not be overridden'.format(self.__class__))
@property
def related(self):
"""Return related providers generator."""
yield from self.__providers.values()
yield from super().related
cpdef object _provide(self, tuple args, dict kwargs):
try:
provider_name = args[0]
except IndexError:
try:
provider_name = kwargs.pop("factory_name")
except KeyError:
raise TypeError("Missing 1st required positional argument: \"provider_name\"")
else:
args = args[1:]
return self.__get_provider(provider_name)(*args, **kwargs)
cdef Provider __get_provider(self, object provider_name):
if provider_name not in self.__providers:
raise NoSuchProviderError("{0} does not contain provider with name {1}".format(self, provider_name))
return <Provider> self.__providers[provider_name]
cdef class Dependency(Provider):
""":py:class:`Dependency` provider describes dependency interface.