mirror of
https://github.com/ets-labs/python-dependency-injector.git
synced 2025-04-13 21:54:24 +03:00
Run black
This commit is contained in:
ZipFile
parent
3893e1df81
commit
50643e0dfb
|
|
@ -19,21 +19,24 @@ from typing import (
|
|||
|
||||
from .providers import Provider, Self, ProviderParent
|
||||
|
||||
|
||||
C_Base = TypeVar("C_Base", bound="Container")
|
||||
C = TypeVar("C", bound="DeclarativeContainer")
|
||||
C_Overriding = TypeVar("C_Overriding", bound="DeclarativeContainer")
|
||||
T = TypeVar("T")
|
||||
TT = TypeVar("TT")
|
||||
|
||||
|
||||
class WiringConfiguration:
|
||||
modules: List[Any]
|
||||
packages: List[Any]
|
||||
from_package: Optional[str]
|
||||
auto_wire: bool
|
||||
def __init__(self, modules: Optional[Iterable[Any]] = None, packages: Optional[Iterable[Any]] = None, from_package: Optional[str] = None, auto_wire: bool = True) -> None: ...
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
modules: Optional[Iterable[Any]] = None,
|
||||
packages: Optional[Iterable[Any]] = None,
|
||||
from_package: Optional[str] = None,
|
||||
auto_wire: bool = True,
|
||||
) -> None: ...
|
||||
|
||||
class Container:
|
||||
provider_type: Type[Provider] = Provider
|
||||
|
|
@ -51,11 +54,18 @@ class Container:
|
|||
def set_providers(self, **providers: Provider): ...
|
||||
def set_provider(self, name: str, provider: Provider) -> None: ...
|
||||
def override(self, overriding: Union[Container, Type[Container]]) -> None: ...
|
||||
def override_providers(self, **overriding_providers: Union[Provider, Any]) -> ProvidersOverridingContext[C_Base]: ...
|
||||
def override_providers(
|
||||
self, **overriding_providers: Union[Provider, Any]
|
||||
) -> ProvidersOverridingContext[C_Base]: ...
|
||||
def reset_last_overriding(self) -> None: ...
|
||||
def reset_override(self) -> None: ...
|
||||
def is_auto_wiring_enabled(self) -> bool: ...
|
||||
def wire(self, modules: Optional[Iterable[Any]] = None, packages: Optional[Iterable[Any]] = None, from_package: Optional[str] = None) -> None: ...
|
||||
def wire(
|
||||
self,
|
||||
modules: Optional[Iterable[Any]] = None,
|
||||
packages: Optional[Iterable[Any]] = None,
|
||||
from_package: Optional[str] = None,
|
||||
) -> None: ...
|
||||
def unwire(self) -> None: ...
|
||||
def init_resources(self) -> Optional[Awaitable]: ...
|
||||
def shutdown_resources(self) -> Optional[Awaitable]: ...
|
||||
|
|
@ -64,7 +74,9 @@ class Container:
|
|||
def reset_singletons(self) -> SingletonResetContext[C_Base]: ...
|
||||
def check_dependencies(self) -> None: ...
|
||||
def from_schema(self, schema: Dict[Any, Any]) -> None: ...
|
||||
def from_yaml_schema(self, filepath: Union[Path, str], loader: Optional[Any]=None) -> None: ...
|
||||
def from_yaml_schema(
|
||||
self, filepath: Union[Path, str], loader: Optional[Any] = None
|
||||
) -> None: ...
|
||||
def from_json_schema(self, filepath: Union[Path, str]) -> None: ...
|
||||
@overload
|
||||
def resolve_provider_name(self, provider: Provider) -> str: ...
|
||||
|
|
@ -82,10 +94,8 @@ class Container:
|
|||
@overload
|
||||
def traverse(cls, types: Optional[Iterable[Type[TT]]] = None) -> Iterator[TT]: ...
|
||||
|
||||
|
||||
class DynamicContainer(Container): ...
|
||||
|
||||
|
||||
class DeclarativeContainer(Container):
|
||||
cls_providers: ClassVar[Dict[str, Provider]]
|
||||
inherited_providers: ClassVar[Dict[str, Provider]]
|
||||
|
|
@ -93,29 +103,28 @@ class DeclarativeContainer(Container):
|
|||
@classmethod
|
||||
def override(cls, overriding: Union[Container, Type[Container]]) -> None: ...
|
||||
@classmethod
|
||||
def override_providers(cls, **overriding_providers: Union[Provider, Any]) -> ProvidersOverridingContext[C_Base]: ...
|
||||
def override_providers(
|
||||
cls, **overriding_providers: Union[Provider, Any]
|
||||
) -> ProvidersOverridingContext[C_Base]: ...
|
||||
@classmethod
|
||||
def reset_last_overriding(cls) -> None: ...
|
||||
@classmethod
|
||||
def reset_override(cls) -> None: ...
|
||||
|
||||
|
||||
class ProvidersOverridingContext(Generic[T]):
|
||||
def __init__(self, container: T, overridden_providers: Iterable[Union[Provider, Any]]) -> None: ...
|
||||
def __init__(
|
||||
self, container: T, overridden_providers: Iterable[Union[Provider, Any]]
|
||||
) -> None: ...
|
||||
def __enter__(self) -> T: ...
|
||||
def __exit__(self, *_: Any) -> None: ...
|
||||
|
||||
|
||||
class SingletonResetContext(Generic[T]):
|
||||
def __init__(self, container: T): ...
|
||||
def __enter__(self) -> T: ...
|
||||
def __exit__(self, *_: Any) -> None: ...
|
||||
|
||||
|
||||
def override(container: Type[C]) -> _Callable[[Type[C_Overriding]], Type[C_Overriding]]: ...
|
||||
|
||||
|
||||
def override(
|
||||
container: Type[C],
|
||||
) -> _Callable[[Type[C_Overriding]], Type[C_Overriding]]: ...
|
||||
def copy(container: Type[C]) -> _Callable[[Type[C_Overriding]], Type[C_Overriding]]: ...
|
||||
|
||||
|
||||
def is_container(instance: Any) -> bool: ...
|
||||
|
|
|
|||
|
|
@ -38,9 +38,11 @@ class View(providers.Callable):
|
|||
|
||||
def as_view(self):
|
||||
"""Return aiohttp view function."""
|
||||
|
||||
@functools.wraps(self.provides)
|
||||
async def _view(request, *args, **kwargs):
|
||||
return await self.__call__(request, *args, **kwargs)
|
||||
|
||||
return _view
|
||||
|
||||
|
||||
|
|
@ -49,6 +51,8 @@ class ClassBasedView(providers.Factory):
|
|||
|
||||
def as_view(self):
|
||||
"""Return aiohttp view function."""
|
||||
|
||||
async def _view(request, *args, **kwargs):
|
||||
return await self.__call__(request, *args, **kwargs)
|
||||
|
||||
return _view
|
||||
|
|
|
|||
|
|
@ -2,22 +2,13 @@ from typing import Awaitable as _Awaitable
|
|||
|
||||
from dependency_injector import providers
|
||||
|
||||
|
||||
class Application(providers.Singleton): ...
|
||||
|
||||
|
||||
class Extension(providers.Singleton): ...
|
||||
|
||||
|
||||
class Middleware(providers.DelegatedCallable): ...
|
||||
|
||||
|
||||
class MiddlewareFactory(providers.Factory): ...
|
||||
|
||||
|
||||
class View(providers.Callable):
|
||||
def as_view(self) -> _Awaitable: ...
|
||||
|
||||
|
||||
class ClassBasedView(providers.Factory):
|
||||
def as_view(self) -> _Awaitable: ...
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ class ClassBasedView(providers.Factory):
|
|||
def as_view(provider, name=None):
|
||||
"""Transform class-based view provider to view function."""
|
||||
if isinstance(provider, providers.Factory):
|
||||
|
||||
def view(*args, **kwargs):
|
||||
self = provider()
|
||||
return self.dispatch_request(*args, **kwargs)
|
||||
|
|
@ -52,12 +53,13 @@ def as_view(provider, name=None):
|
|||
assert name, 'Argument "endpoint" is required for class-based views'
|
||||
view.__name__ = name
|
||||
elif isinstance(provider, providers.Callable):
|
||||
|
||||
def view(*args, **kwargs):
|
||||
return provider(*args, **kwargs)
|
||||
|
||||
view.__name__ = provider.provides.__name__
|
||||
else:
|
||||
raise errors.Error('Undefined provider type')
|
||||
raise errors.Error("Undefined provider type")
|
||||
|
||||
view.__doc__ = provider.provides.__doc__
|
||||
view.__module__ = provider.provides.__module__
|
||||
|
|
@ -65,14 +67,14 @@ def as_view(provider, name=None):
|
|||
if isinstance(provider.provides, type):
|
||||
view.view_class = provider.provides
|
||||
|
||||
if hasattr(provider.provides, 'decorators'):
|
||||
if hasattr(provider.provides, "decorators"):
|
||||
for decorator in provider.provides.decorators:
|
||||
view = decorator(view)
|
||||
|
||||
if hasattr(provider.provides, 'methods'):
|
||||
if hasattr(provider.provides, "methods"):
|
||||
view.methods = provider.provides.methods
|
||||
|
||||
if hasattr(provider.provides, 'provide_automatic_options'):
|
||||
if hasattr(provider.provides, "provide_automatic_options"):
|
||||
view.provide_automatic_options = provider.provides.provide_automatic_options
|
||||
|
||||
return view
|
||||
|
|
|
|||
|
|
@ -3,22 +3,17 @@ from typing import Union, Optional, Callable as _Callable, Any
|
|||
from flask import request as flask_request
|
||||
from dependency_injector import providers
|
||||
|
||||
|
||||
request: providers.Object[flask_request]
|
||||
|
||||
|
||||
class Application(providers.Singleton): ...
|
||||
|
||||
|
||||
class Extension(providers.Singleton): ...
|
||||
|
||||
|
||||
class View(providers.Callable):
|
||||
def as_view(self) -> _Callable[..., Any]: ...
|
||||
|
||||
|
||||
class ClassBasedView(providers.Factory):
|
||||
def as_view(self, name: str) -> _Callable[..., Any]: ...
|
||||
|
||||
|
||||
def as_view(provider: Union[View, ClassBasedView], name: Optional[str] = None) -> _Callable[..., Any]: ...
|
||||
def as_view(
|
||||
provider: Union[View, ClassBasedView], name: Optional[str] = None
|
||||
) -> _Callable[..., Any]: ...
|
||||
|
|
|
|||
|
|
@ -33,7 +33,6 @@ except ImportError:
|
|||
|
||||
from . import resources
|
||||
|
||||
|
||||
Injection = Any
|
||||
ProviderParent = Union["Provider", Any]
|
||||
T = TypeVar("T")
|
||||
|
|
@ -41,16 +40,13 @@ TT = TypeVar("TT")
|
|||
P = TypeVar("P", bound="Provider")
|
||||
BS = TypeVar("BS", bound="BaseSingleton")
|
||||
|
||||
|
||||
class Provider(Generic[T]):
|
||||
def __init__(self) -> None: ...
|
||||
|
||||
@overload
|
||||
def __call__(self, *args: Injection, **kwargs: Injection) -> T: ...
|
||||
@overload
|
||||
def __call__(self, *args: Injection, **kwargs: Injection) -> Awaitable[T]: ...
|
||||
def async_(self, *args: Injection, **kwargs: Injection) -> Awaitable[T]: ...
|
||||
|
||||
def __deepcopy__(self, memo: Optional[_Dict[Any, Any]]) -> Provider: ...
|
||||
def __str__(self) -> str: ...
|
||||
def __repr__(self) -> str: ...
|
||||
|
|
@ -78,9 +74,12 @@ class Provider(Generic[T]):
|
|||
def is_async_mode_undefined(self) -> bool: ...
|
||||
@property
|
||||
def related(self) -> _Iterator[Provider]: ...
|
||||
def traverse(self, types: Optional[_Iterable[Type[TT]]] = None) -> _Iterator[TT]: ...
|
||||
def _copy_overridings(self, copied: Provider, memo: Optional[_Dict[Any, Any]]) -> None: ...
|
||||
|
||||
def traverse(
|
||||
self, types: Optional[_Iterable[Type[TT]]] = None
|
||||
) -> _Iterator[TT]: ...
|
||||
def _copy_overridings(
|
||||
self, copied: Provider, memo: Optional[_Dict[Any, Any]]
|
||||
) -> None: ...
|
||||
|
||||
class Object(Provider[T]):
|
||||
def __init__(self, provides: Optional[T] = None) -> None: ...
|
||||
|
|
@ -88,7 +87,6 @@ class Object(Provider[T]):
|
|||
def provides(self) -> Optional[T]: ...
|
||||
def set_provides(self, provides: Optional[T]) -> Object: ...
|
||||
|
||||
|
||||
class Self(Provider[T]):
|
||||
def __init__(self, container: Optional[T] = None) -> None: ...
|
||||
def set_container(self, container: T) -> None: ...
|
||||
|
|
@ -96,41 +94,51 @@ class Self(Provider[T]):
|
|||
@property
|
||||
def alt_names(self) -> Tuple[Any]: ...
|
||||
|
||||
|
||||
class Delegate(Provider[Provider]):
|
||||
def __init__(self, provides: Optional[Provider] = None) -> None: ...
|
||||
@property
|
||||
def provides(self) -> Optional[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 __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: ...
|
||||
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]: ...
|
||||
|
||||
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]: ...
|
||||
|
||||
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 __init__(
|
||||
self,
|
||||
instance_of: Type[T] = object,
|
||||
default: Optional[Union[Provider, Any]] = None,
|
||||
) -> None: ...
|
||||
def __getattr__(self, name: str) -> Any: ...
|
||||
|
||||
@property
|
||||
def instance_of(self) -> Type[T]: ...
|
||||
def set_instance_of(self, instance_of: Type[T]) -> Dependency[T]: ...
|
||||
|
||||
@property
|
||||
def default(self) -> Provider[T]: ...
|
||||
def set_default(self, default: Optional[Union[Provider, Any]]) -> Dependency[T]: ...
|
||||
|
||||
@property
|
||||
def is_defined(self) -> bool: ...
|
||||
def provided_by(self, provider: Provider) -> OverridingContext[P]: ...
|
||||
|
|
@ -140,10 +148,8 @@ class Dependency(Provider[T]):
|
|||
def parent_name(self) -> Optional[str]: ...
|
||||
def assign_parent(self, parent: ProviderParent) -> None: ...
|
||||
|
||||
|
||||
class ExternalDependency(Dependency[T]): ...
|
||||
|
||||
|
||||
class DependenciesContainer(Object):
|
||||
def __init__(self, **dependencies: Provider) -> None: ...
|
||||
def __getattr__(self, name: str) -> Provider: ...
|
||||
|
|
@ -156,12 +162,18 @@ class DependenciesContainer(Object):
|
|||
def parent_name(self) -> Optional[str]: ...
|
||||
def assign_parent(self, parent: ProviderParent) -> None: ...
|
||||
|
||||
|
||||
class Callable(Provider[T]):
|
||||
def __init__(self, provides: Optional[Union[_Callable[..., T], str]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[Union[_Callable[..., T], str]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@property
|
||||
def provides(self) -> Optional[_Callable[..., T]]: ...
|
||||
def set_provides(self, provides: Optional[Union[_Callable[..., T], str]]) -> Callable[T]: ...
|
||||
def set_provides(
|
||||
self, provides: Optional[Union[_Callable[..., T], str]]
|
||||
) -> Callable[T]: ...
|
||||
@property
|
||||
def args(self) -> Tuple[Injection]: ...
|
||||
def add_args(self, *args: Injection) -> Callable[T]: ...
|
||||
|
|
@ -173,32 +185,23 @@ class Callable(Provider[T]):
|
|||
def set_kwargs(self, **kwargs: Injection) -> Callable[T]: ...
|
||||
def clear_kwargs(self) -> Callable[T]: ...
|
||||
|
||||
|
||||
class DelegatedCallable(Callable[T]): ...
|
||||
|
||||
|
||||
class AbstractCallable(Callable[T]):
|
||||
def override(self, provider: Callable) -> OverridingContext[P]: ...
|
||||
|
||||
|
||||
class CallableDelegate(Delegate):
|
||||
def __init__(self, callable: Callable) -> None: ...
|
||||
|
||||
|
||||
class Coroutine(Callable[T]): ...
|
||||
|
||||
|
||||
class DelegatedCoroutine(Coroutine[T]): ...
|
||||
|
||||
|
||||
class AbstractCoroutine(Coroutine[T]):
|
||||
def override(self, provider: Coroutine) -> OverridingContext[P]: ...
|
||||
|
||||
|
||||
class CoroutineDelegate(Delegate):
|
||||
def __init__(self, coroutine: Coroutine) -> None: ...
|
||||
|
||||
|
||||
class ConfigurationOption(Provider[Any]):
|
||||
UNDEFINED: object
|
||||
def __init__(self, name: Tuple[str], root: Configuration) -> None: ...
|
||||
|
|
@ -212,89 +215,137 @@ class ConfigurationOption(Provider[Any]):
|
|||
def get_name_segments(self) -> Tuple[Union[str, Provider]]: ...
|
||||
def as_int(self) -> TypedConfigurationOption[int]: ...
|
||||
def as_float(self) -> TypedConfigurationOption[float]: ...
|
||||
def as_(self, callback: _Callable[..., T], *args: Injection, **kwargs: Injection) -> TypedConfigurationOption[T]: ...
|
||||
def as_(
|
||||
self, callback: _Callable[..., T], *args: Injection, **kwargs: Injection
|
||||
) -> TypedConfigurationOption[T]: ...
|
||||
def required(self) -> ConfigurationOption: ...
|
||||
def is_required(self) -> bool: ...
|
||||
def update(self, value: Any) -> None: ...
|
||||
def from_ini(self, filepath: Union[Path, str], required: bool = False, envs_required: bool = False) -> None: ...
|
||||
def from_yaml(self, filepath: Union[Path, str], required: bool = False, loader: Optional[Any] = None, envs_required: bool = False) -> None: ...
|
||||
def from_json(self, filepath: Union[Path, str], required: bool = False, envs_required: bool = False) -> None: ...
|
||||
def from_pydantic(self, settings: PydanticSettings, required: bool = False, **kwargs: Any) -> None: ...
|
||||
def from_ini(
|
||||
self,
|
||||
filepath: Union[Path, str],
|
||||
required: bool = False,
|
||||
envs_required: bool = False,
|
||||
) -> None: ...
|
||||
def from_yaml(
|
||||
self,
|
||||
filepath: Union[Path, str],
|
||||
required: bool = False,
|
||||
loader: Optional[Any] = None,
|
||||
envs_required: bool = False,
|
||||
) -> None: ...
|
||||
def from_json(
|
||||
self,
|
||||
filepath: Union[Path, str],
|
||||
required: bool = False,
|
||||
envs_required: bool = False,
|
||||
) -> None: ...
|
||||
def from_pydantic(
|
||||
self, settings: PydanticSettings, required: bool = False, **kwargs: Any
|
||||
) -> None: ...
|
||||
def from_dict(self, options: _Dict[str, Any], required: bool = False) -> None: ...
|
||||
def from_env(self, name: str, default: Optional[Any] = None, required: bool = False, as_: Optional[_Callable[..., Any]] = None) -> None: ...
|
||||
def from_env(
|
||||
self,
|
||||
name: str,
|
||||
default: Optional[Any] = None,
|
||||
required: bool = False,
|
||||
as_: Optional[_Callable[..., Any]] = None,
|
||||
) -> None: ...
|
||||
def from_value(self, value: Any) -> None: ...
|
||||
|
||||
|
||||
class TypedConfigurationOption(Callable[T]):
|
||||
@property
|
||||
def option(self) -> ConfigurationOption: ...
|
||||
|
||||
|
||||
class Configuration(Object[Any]):
|
||||
DEFAULT_NAME: str = "config"
|
||||
def __init__(
|
||||
self,
|
||||
name: str = DEFAULT_NAME,
|
||||
default: Optional[Any] = None,
|
||||
*,
|
||||
strict: bool = False,
|
||||
ini_files: Optional[_Iterable[Union[Path, str]]] = None,
|
||||
yaml_files: Optional[_Iterable[Union[Path, str]]] = None,
|
||||
json_files: Optional[_Iterable[Union[Path, str]]] = None,
|
||||
pydantic_settings: Optional[_Iterable[PydanticSettings]] = None,
|
||||
self,
|
||||
name: str = DEFAULT_NAME,
|
||||
default: Optional[Any] = None,
|
||||
*,
|
||||
strict: bool = False,
|
||||
ini_files: Optional[_Iterable[Union[Path, str]]] = None,
|
||||
yaml_files: Optional[_Iterable[Union[Path, str]]] = None,
|
||||
json_files: Optional[_Iterable[Union[Path, str]]] = None,
|
||||
pydantic_settings: Optional[_Iterable[PydanticSettings]] = None,
|
||||
) -> None: ...
|
||||
def __enter__(self) -> Configuration : ...
|
||||
def __enter__(self) -> Configuration: ...
|
||||
def __exit__(self, *exc_info: Any) -> None: ...
|
||||
def __getattr__(self, item: str) -> ConfigurationOption: ...
|
||||
def __getitem__(self, item: Union[str, Provider]) -> ConfigurationOption: ...
|
||||
|
||||
def get_name(self) -> str: ...
|
||||
def set_name(self, name: str) -> Configuration: ...
|
||||
|
||||
def get_default(self) -> _Dict[Any, Any]: ...
|
||||
def set_default(self, default: _Dict[Any, Any]): ...
|
||||
|
||||
def get_strict(self) -> bool: ...
|
||||
def set_strict(self, strict: bool) -> Configuration: ...
|
||||
|
||||
def get_children(self) -> _Dict[str, ConfigurationOption]: ...
|
||||
def set_children(self, children: _Dict[str, ConfigurationOption]) -> Configuration: ...
|
||||
|
||||
def set_children(
|
||||
self, children: _Dict[str, ConfigurationOption]
|
||||
) -> Configuration: ...
|
||||
def get_ini_files(self) -> _List[Union[Path, str]]: ...
|
||||
def set_ini_files(self, files: _Iterable[Union[Path, str]]) -> Configuration: ...
|
||||
|
||||
def get_yaml_files(self) -> _List[Union[Path, str]]: ...
|
||||
def set_yaml_files(self, files: _Iterable[Union[Path, str]]) -> Configuration: ...
|
||||
|
||||
def get_json_files(self) -> _List[Union[Path, str]]: ...
|
||||
def set_json_files(self, files: _Iterable[Union[Path, str]]) -> Configuration: ...
|
||||
|
||||
def get_pydantic_settings(self) -> _List[PydanticSettings]: ...
|
||||
def set_pydantic_settings(self, settings: _Iterable[PydanticSettings]) -> Configuration: ...
|
||||
|
||||
def set_pydantic_settings(
|
||||
self, settings: _Iterable[PydanticSettings]
|
||||
) -> Configuration: ...
|
||||
def load(self, required: bool = False, envs_required: bool = False) -> None: ...
|
||||
|
||||
def get(self, selector: str) -> Any: ...
|
||||
def set(self, selector: str, value: Any) -> OverridingContext[P]: ...
|
||||
def reset_cache(self) -> None: ...
|
||||
def update(self, value: Any) -> None: ...
|
||||
def from_ini(self, filepath: Union[Path, str], required: bool = False, envs_required: bool = False) -> None: ...
|
||||
def from_yaml(self, filepath: Union[Path, str], required: bool = False, loader: Optional[Any] = None, envs_required: bool = False) -> None: ...
|
||||
def from_json(self, filepath: Union[Path, str], required: bool = False, envs_required: bool = False) -> None: ...
|
||||
def from_pydantic(self, settings: PydanticSettings, required: bool = False, **kwargs: Any) -> None: ...
|
||||
def from_ini(
|
||||
self,
|
||||
filepath: Union[Path, str],
|
||||
required: bool = False,
|
||||
envs_required: bool = False,
|
||||
) -> None: ...
|
||||
def from_yaml(
|
||||
self,
|
||||
filepath: Union[Path, str],
|
||||
required: bool = False,
|
||||
loader: Optional[Any] = None,
|
||||
envs_required: bool = False,
|
||||
) -> None: ...
|
||||
def from_json(
|
||||
self,
|
||||
filepath: Union[Path, str],
|
||||
required: bool = False,
|
||||
envs_required: bool = False,
|
||||
) -> None: ...
|
||||
def from_pydantic(
|
||||
self, settings: PydanticSettings, required: bool = False, **kwargs: Any
|
||||
) -> None: ...
|
||||
def from_dict(self, options: _Dict[str, Any], required: bool = False) -> None: ...
|
||||
def from_env(self, name: str, default: Optional[Any] = None, required: bool = False, as_: Optional[_Callable[..., Any]] = None) -> None: ...
|
||||
def from_env(
|
||||
self,
|
||||
name: str,
|
||||
default: Optional[Any] = None,
|
||||
required: bool = False,
|
||||
as_: Optional[_Callable[..., Any]] = None,
|
||||
) -> None: ...
|
||||
def from_value(self, value: Any) -> None: ...
|
||||
|
||||
|
||||
class Factory(Provider[T]):
|
||||
provided_type: Optional[Type]
|
||||
def __init__(self, provides: Optional[Union[_Callable[..., T], str]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[Union[_Callable[..., T], str]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@property
|
||||
def cls(self) -> Type[T]: ...
|
||||
@property
|
||||
def provides(self) -> Optional[_Callable[..., T]]: ...
|
||||
def set_provides(self, provides: Optional[Union[_Callable[..., T], str]]) -> Factory[T]: ...
|
||||
def set_provides(
|
||||
self, provides: Optional[Union[_Callable[..., T], str]]
|
||||
) -> Factory[T]: ...
|
||||
@property
|
||||
def args(self) -> Tuple[Injection]: ...
|
||||
def add_args(self, *args: Injection) -> Factory[T]: ...
|
||||
|
|
@ -311,33 +362,39 @@ class Factory(Provider[T]):
|
|||
def set_attributes(self, **kwargs: Injection) -> Factory[T]: ...
|
||||
def clear_attributes(self) -> Factory[T]: ...
|
||||
|
||||
|
||||
class DelegatedFactory(Factory[T]): ...
|
||||
|
||||
|
||||
class AbstractFactory(Factory[T]):
|
||||
def override(self, provider: Factory) -> OverridingContext[P]: ...
|
||||
|
||||
|
||||
class FactoryDelegate(Delegate):
|
||||
def __init__(self, factory: Factory): ...
|
||||
|
||||
|
||||
class FactoryAggregate(Aggregate[T]):
|
||||
def __getattr__(self, provider_name: Any) -> Factory[T]: ...
|
||||
@property
|
||||
def factories(self) -> _Dict[Any, Factory[T]]: ...
|
||||
def set_factories(self, provider_dict: Optional[_Dict[Any, Factory[T]]] = None, **provider_kwargs: Factory[T]) -> FactoryAggregate[T]: ...
|
||||
|
||||
def set_factories(
|
||||
self,
|
||||
provider_dict: Optional[_Dict[Any, Factory[T]]] = None,
|
||||
**provider_kwargs: Factory[T],
|
||||
) -> FactoryAggregate[T]: ...
|
||||
|
||||
class BaseSingleton(Provider[T]):
|
||||
provided_type = Optional[Type]
|
||||
def __init__(self, provides: Optional[Union[_Callable[..., T], str]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[Union[_Callable[..., T], str]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@property
|
||||
def cls(self) -> Type[T]: ...
|
||||
@property
|
||||
def provides(self) -> Optional[_Callable[..., T]]: ...
|
||||
def set_provides(self, provides: Optional[Union[_Callable[..., T], str]]) -> BaseSingleton[T]: ...
|
||||
def set_provides(
|
||||
self, provides: Optional[Union[_Callable[..., T], str]]
|
||||
) -> BaseSingleton[T]: ...
|
||||
@property
|
||||
def args(self) -> Tuple[Injection]: ...
|
||||
def add_args(self, *args: Injection) -> BaseSingleton[T]: ...
|
||||
|
|
@ -356,36 +413,20 @@ class BaseSingleton(Provider[T]):
|
|||
def reset(self) -> SingletonResetContext[BS]: ...
|
||||
def full_reset(self) -> SingletonFullResetContext[BS]: ...
|
||||
|
||||
|
||||
class Singleton(BaseSingleton[T]): ...
|
||||
|
||||
|
||||
class DelegatedSingleton(Singleton[T]): ...
|
||||
|
||||
|
||||
class ThreadSafeSingleton(Singleton[T]): ...
|
||||
|
||||
|
||||
class DelegatedThreadSafeSingleton(ThreadSafeSingleton[T]): ...
|
||||
|
||||
|
||||
class ThreadLocalSingleton(BaseSingleton[T]): ...
|
||||
|
||||
|
||||
class ContextLocalSingleton(BaseSingleton[T]): ...
|
||||
|
||||
|
||||
class DelegatedThreadLocalSingleton(ThreadLocalSingleton[T]): ...
|
||||
|
||||
|
||||
class AbstractSingleton(BaseSingleton[T]):
|
||||
def override(self, provider: BaseSingleton) -> OverridingContext[P]: ...
|
||||
|
||||
|
||||
class SingletonDelegate(Delegate):
|
||||
def __init__(self, singleton: BaseSingleton): ...
|
||||
|
||||
|
||||
class List(Provider[_List]):
|
||||
def __init__(self, *args: Injection): ...
|
||||
@property
|
||||
|
|
@ -394,29 +435,63 @@ class List(Provider[_List]):
|
|||
def set_args(self, *args: Injection) -> List[T]: ...
|
||||
def clear_args(self) -> List[T]: ...
|
||||
|
||||
|
||||
class Dict(Provider[_Dict]):
|
||||
def __init__(self, dict_: Optional[_Dict[Any, Injection]] = None, **kwargs: Injection): ...
|
||||
def __init__(
|
||||
self, dict_: Optional[_Dict[Any, Injection]] = None, **kwargs: Injection
|
||||
): ...
|
||||
@property
|
||||
def kwargs(self) -> _Dict[Any, Injection]: ...
|
||||
def add_kwargs(self, dict_: Optional[_Dict[Any, Injection]] = None, **kwargs: Injection) -> Dict: ...
|
||||
def set_kwargs(self, dict_: Optional[_Dict[Any, Injection]] = None, **kwargs: Injection) -> Dict: ...
|
||||
def add_kwargs(
|
||||
self, dict_: Optional[_Dict[Any, Injection]] = None, **kwargs: Injection
|
||||
) -> Dict: ...
|
||||
def set_kwargs(
|
||||
self, dict_: Optional[_Dict[Any, Injection]] = None, **kwargs: Injection
|
||||
) -> Dict: ...
|
||||
def clear_kwargs(self) -> Dict: ...
|
||||
|
||||
|
||||
class Resource(Provider[T]):
|
||||
@overload
|
||||
def __init__(self, provides: Optional[Type[resources.Resource[T]]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[Type[resources.Resource[T]]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@overload
|
||||
def __init__(self, provides: Optional[Type[resources.AsyncResource[T]]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[Type[resources.AsyncResource[T]]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@overload
|
||||
def __init__(self, provides: Optional[_Callable[..., _Iterator[T]]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[_Callable[..., _Iterator[T]]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@overload
|
||||
def __init__(self, provides: Optional[_Callable[..., _AsyncIterator[T]]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[_Callable[..., _AsyncIterator[T]]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@overload
|
||||
def __init__(self, provides: Optional[_Callable[..., _Coroutine[Injection, Injection, T]]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[_Callable[..., _Coroutine[Injection, Injection, T]]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@overload
|
||||
def __init__(self, provides: Optional[Union[_Callable[..., T], str]] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
provides: Optional[Union[_Callable[..., T], str]] = None,
|
||||
*args: Injection,
|
||||
**kwargs: Injection,
|
||||
) -> None: ...
|
||||
@property
|
||||
def provides(self) -> Optional[_Callable[..., Any]]: ...
|
||||
def set_provides(self, provides: Optional[Any]) -> Resource[T]: ...
|
||||
|
|
@ -435,9 +510,13 @@ class Resource(Provider[T]):
|
|||
def init(self) -> Optional[Awaitable[T]]: ...
|
||||
def shutdown(self) -> Optional[Awaitable]: ...
|
||||
|
||||
|
||||
class Container(Provider[T]):
|
||||
def __init__(self, container_cls: Type[T], container: Optional[T] = None, **overriding_providers: Union[Provider, Any]) -> None: ...
|
||||
def __init__(
|
||||
self,
|
||||
container_cls: Type[T],
|
||||
container: Optional[T] = None,
|
||||
**overriding_providers: Union[Provider, Any],
|
||||
) -> None: ...
|
||||
def __getattr__(self, name: str) -> Provider: ...
|
||||
@property
|
||||
def container(self) -> T: ...
|
||||
|
|
@ -448,50 +527,51 @@ class Container(Provider[T]):
|
|||
def parent_name(self) -> Optional[str]: ...
|
||||
def assign_parent(self, parent: ProviderParent) -> None: ...
|
||||
|
||||
|
||||
class Selector(Provider[Any]):
|
||||
def __init__(self, selector: Optional[_Callable[..., Any]] = None, **providers: Provider): ...
|
||||
def __init__(
|
||||
self, selector: Optional[_Callable[..., Any]] = None, **providers: Provider
|
||||
): ...
|
||||
def __getattr__(self, name: str) -> Provider: ...
|
||||
|
||||
@property
|
||||
def selector(self) -> Optional[_Callable[..., Any]]: ...
|
||||
def set_selector(self, selector: Optional[_Callable[..., Any]]) -> Selector: ...
|
||||
|
||||
@property
|
||||
def providers(self) -> _Dict[str, Provider]: ...
|
||||
def set_providers(self, **providers: Provider) -> Selector: ...
|
||||
|
||||
|
||||
class ProvidedInstanceFluentInterface:
|
||||
def __getattr__(self, item: Any) -> AttributeGetter: ...
|
||||
def __getitem__(self, item: Any) -> ItemGetter: ...
|
||||
def call(self, *args: Injection, **kwargs: Injection) -> MethodCaller: ...
|
||||
@property
|
||||
def provides(self) -> Optional[Provider]: ...
|
||||
def set_provides(self, provides: Optional[Provider]) -> ProvidedInstanceFluentInterface: ...
|
||||
|
||||
def set_provides(
|
||||
self, provides: Optional[Provider]
|
||||
) -> ProvidedInstanceFluentInterface: ...
|
||||
|
||||
class ProvidedInstance(Provider, ProvidedInstanceFluentInterface):
|
||||
def __init__(self, provides: Optional[Provider] = None) -> None: ...
|
||||
|
||||
|
||||
class AttributeGetter(Provider, ProvidedInstanceFluentInterface):
|
||||
def __init__(self, provides: Optional[Provider] = None, name: Optional[str] = None) -> None: ...
|
||||
def __init__(
|
||||
self, provides: Optional[Provider] = None, name: Optional[str] = None
|
||||
) -> None: ...
|
||||
@property
|
||||
def name(self) -> Optional[str]: ...
|
||||
def set_name(self, name: Optional[str]) -> ProvidedInstanceFluentInterface: ...
|
||||
|
||||
|
||||
class ItemGetter(Provider, ProvidedInstanceFluentInterface):
|
||||
def __init__(self, provides: Optional[Provider] = None, name: Optional[str] = None) -> None: ...
|
||||
def __init__(
|
||||
self, provides: Optional[Provider] = None, name: Optional[str] = None
|
||||
) -> None: ...
|
||||
@property
|
||||
def name(self) -> Optional[str]: ...
|
||||
def set_name(self, name: Optional[str]) -> ProvidedInstanceFluentInterface: ...
|
||||
|
||||
|
||||
class MethodCaller(Provider, ProvidedInstanceFluentInterface):
|
||||
def __init__(self, provides: Optional[Provider] = None, *args: Injection, **kwargs: Injection) -> None: ...
|
||||
|
||||
def __init__(
|
||||
self, provides: Optional[Provider] = None, *args: Injection, **kwargs: Injection
|
||||
) -> None: ...
|
||||
|
||||
class OverridingContext(Generic[T]):
|
||||
def __init__(self, overridden: Provider, overriding: Provider): ...
|
||||
|
|
@ -500,61 +580,39 @@ class OverridingContext(Generic[T]):
|
|||
pass
|
||||
...
|
||||
|
||||
|
||||
class BaseSingletonResetContext(Generic[T]):
|
||||
def __init__(self, provider: T): ...
|
||||
def __enter__(self) -> T: ...
|
||||
def __exit__(self, *_: Any) -> None: ...
|
||||
|
||||
|
||||
class SingletonResetContext(BaseSingletonResetContext):
|
||||
...
|
||||
|
||||
|
||||
class SingletonFullResetContext(BaseSingletonResetContext):
|
||||
...
|
||||
|
||||
class SingletonResetContext(BaseSingletonResetContext): ...
|
||||
class SingletonFullResetContext(BaseSingletonResetContext): ...
|
||||
|
||||
CHILD_PROVIDERS: Tuple[Provider]
|
||||
|
||||
|
||||
def is_provider(instance: Any) -> bool: ...
|
||||
|
||||
|
||||
def ensure_is_provider(instance: Any) -> Provider: ...
|
||||
|
||||
|
||||
def is_delegated(instance: Any) -> bool: ...
|
||||
|
||||
|
||||
def represent_provider(provider: Provider, provides: Any) -> str: ...
|
||||
|
||||
|
||||
def deepcopy(instance: Any, memo: Optional[_Dict[Any, Any]] = None) -> Any: ...
|
||||
|
||||
|
||||
def deepcopy_args(
|
||||
provider: Provider[Any],
|
||||
args: Tuple[Any, ...],
|
||||
memo: Optional[_Dict[int, Any]] = None,
|
||||
) -> Tuple[Any, ...]: ...
|
||||
|
||||
|
||||
def deepcopy_kwargs(
|
||||
provider: Provider[Any],
|
||||
kwargs: _Dict[str, Any],
|
||||
memo: Optional[_Dict[int, Any]] = None,
|
||||
) -> Dict[str, Any]: ...
|
||||
|
||||
|
||||
def merge_dicts(dict1: _Dict[Any, Any], dict2: _Dict[Any, Any]) -> _Dict[Any, Any]: ...
|
||||
|
||||
|
||||
def traverse(*providers: Provider, types: Optional[_Iterable[Type]]=None) -> _Iterator[Provider]: ...
|
||||
|
||||
def traverse(
|
||||
*providers: Provider, types: Optional[_Iterable[Type]] = None
|
||||
) -> _Iterator[Provider]: ...
|
||||
|
||||
if yaml:
|
||||
class YamlLoader(yaml.SafeLoader): ...
|
||||
|
||||
else:
|
||||
class YamlLoader: ...
|
||||
|
||||
|
|
|
|||
|
|
@ -10,18 +10,14 @@ T = TypeVar("T")
|
|||
class Resource(Generic[T], metaclass=abc.ABCMeta):
|
||||
|
||||
@abc.abstractmethod
|
||||
def init(self, *args, **kwargs) -> Optional[T]:
|
||||
...
|
||||
def init(self, *args, **kwargs) -> Optional[T]: ...
|
||||
|
||||
def shutdown(self, resource: Optional[T]) -> None:
|
||||
...
|
||||
def shutdown(self, resource: Optional[T]) -> None: ...
|
||||
|
||||
|
||||
class AsyncResource(Generic[T], metaclass=abc.ABCMeta):
|
||||
|
||||
@abc.abstractmethod
|
||||
async def init(self, *args, **kwargs) -> Optional[T]:
|
||||
...
|
||||
async def init(self, *args, **kwargs) -> Optional[T]: ...
|
||||
|
||||
async def shutdown(self, resource: Optional[T]) -> None:
|
||||
...
|
||||
async def shutdown(self, resource: Optional[T]) -> None: ...
|
||||
|
|
|
|||
|
|
@ -27,9 +27,9 @@ class SchemaProcessorV1:
|
|||
return self._container.providers
|
||||
|
||||
def _create_providers(
|
||||
self,
|
||||
provider_schema: ProviderSchema,
|
||||
container: Optional[containers.Container] = None,
|
||||
self,
|
||||
provider_schema: ProviderSchema,
|
||||
container: Optional[containers.Container] = None,
|
||||
) -> None:
|
||||
if container is None:
|
||||
container = self._container
|
||||
|
|
@ -57,9 +57,9 @@ class SchemaProcessorV1:
|
|||
self._create_providers(provider_schema=data, container=provider)
|
||||
|
||||
def _setup_injections( # noqa: C901
|
||||
self,
|
||||
provider_schema: ProviderSchema,
|
||||
container: Optional[containers.Container] = None,
|
||||
self,
|
||||
provider_schema: ProviderSchema,
|
||||
container: Optional[containers.Container] = None,
|
||||
) -> None:
|
||||
if container is None:
|
||||
container = self._container
|
||||
|
|
@ -72,7 +72,7 @@ class SchemaProcessorV1:
|
|||
provides = data.get("provides")
|
||||
if provides:
|
||||
if isinstance(provides, str) and provides.startswith("container."):
|
||||
provides = self._resolve_provider(provides[len("container."):])
|
||||
provides = self._resolve_provider(provides[len("container.") :])
|
||||
else:
|
||||
provides = _import_string(provides)
|
||||
provider.set_provides(provides)
|
||||
|
|
@ -83,7 +83,7 @@ class SchemaProcessorV1:
|
|||
injection = None
|
||||
|
||||
if isinstance(arg, str) and arg.startswith("container."):
|
||||
injection = self._resolve_provider(arg[len("container."):])
|
||||
injection = self._resolve_provider(arg[len("container.") :])
|
||||
|
||||
# TODO: refactoring
|
||||
if isinstance(arg, dict):
|
||||
|
|
@ -91,16 +91,23 @@ class SchemaProcessorV1:
|
|||
provider_type = _get_provider_cls(arg.get("provider"))
|
||||
provides = arg.get("provides")
|
||||
if provides:
|
||||
if isinstance(provides, str) and provides.startswith("container."):
|
||||
provides = self._resolve_provider(provides[len("container."):])
|
||||
if isinstance(provides, str) and provides.startswith(
|
||||
"container."
|
||||
):
|
||||
provides = self._resolve_provider(
|
||||
provides[len("container.") :]
|
||||
)
|
||||
else:
|
||||
provides = _import_string(provides)
|
||||
provider_args.append(provides)
|
||||
for provider_arg in arg.get("args", []):
|
||||
if isinstance(provider_arg, str) \
|
||||
and provider_arg.startswith("container."):
|
||||
if isinstance(
|
||||
provider_arg, str
|
||||
) and provider_arg.startswith("container."):
|
||||
provider_args.append(
|
||||
self._resolve_provider(provider_arg[len("container."):]),
|
||||
self._resolve_provider(
|
||||
provider_arg[len("container.") :]
|
||||
),
|
||||
)
|
||||
injection = provider_type(*provider_args)
|
||||
|
||||
|
|
@ -117,7 +124,7 @@ class SchemaProcessorV1:
|
|||
injection = None
|
||||
|
||||
if isinstance(arg, str) and arg.startswith("container."):
|
||||
injection = self._resolve_provider(arg[len("container."):])
|
||||
injection = self._resolve_provider(arg[len("container.") :])
|
||||
|
||||
# TODO: refactoring
|
||||
if isinstance(arg, dict):
|
||||
|
|
@ -125,16 +132,23 @@ class SchemaProcessorV1:
|
|||
provider_type = _get_provider_cls(arg.get("provider"))
|
||||
provides = arg.get("provides")
|
||||
if provides:
|
||||
if isinstance(provides, str) and provides.startswith("container."):
|
||||
provides = self._resolve_provider(provides[len("container."):])
|
||||
if isinstance(provides, str) and provides.startswith(
|
||||
"container."
|
||||
):
|
||||
provides = self._resolve_provider(
|
||||
provides[len("container.") :]
|
||||
)
|
||||
else:
|
||||
provides = _import_string(provides)
|
||||
provider_args.append(provides)
|
||||
for provider_arg in arg.get("args", []):
|
||||
if isinstance(provider_arg, str) \
|
||||
and provider_arg.startswith("container."):
|
||||
if isinstance(
|
||||
provider_arg, str
|
||||
) and provider_arg.startswith("container."):
|
||||
provider_args.append(
|
||||
self._resolve_provider(provider_arg[len("container."):]),
|
||||
self._resolve_provider(
|
||||
provider_arg[len("container.") :]
|
||||
),
|
||||
)
|
||||
injection = provider_type(*provider_args)
|
||||
|
||||
|
|
@ -158,7 +172,7 @@ class SchemaProcessorV1:
|
|||
for segment in segments[1:]:
|
||||
parentheses = ""
|
||||
if "(" in segment and ")" in segment:
|
||||
parentheses = segment[segment.find("("):segment.rfind(")")+1]
|
||||
parentheses = segment[segment.find("(") : segment.rfind(")") + 1]
|
||||
segment = segment.replace(parentheses, "")
|
||||
|
||||
try:
|
||||
|
|
@ -190,10 +204,12 @@ def _get_provider_cls(provider_cls_name: str) -> Type[providers.Provider]:
|
|||
if custom_provider_type:
|
||||
return custom_provider_type
|
||||
|
||||
raise SchemaError(f"Undefined provider class \"{provider_cls_name}\"")
|
||||
raise SchemaError(f'Undefined provider class "{provider_cls_name}"')
|
||||
|
||||
|
||||
def _fetch_provider_cls_from_std(provider_cls_name: str) -> Optional[Type[providers.Provider]]:
|
||||
def _fetch_provider_cls_from_std(
|
||||
provider_cls_name: str,
|
||||
) -> Optional[Type[providers.Provider]]:
|
||||
return getattr(providers, provider_cls_name, None)
|
||||
|
||||
|
||||
|
|
@ -201,12 +217,16 @@ def _import_provider_cls(provider_cls_name: str) -> Optional[Type[providers.Prov
|
|||
try:
|
||||
cls = _import_string(provider_cls_name)
|
||||
except (ImportError, ValueError) as exception:
|
||||
raise SchemaError(f"Can not import provider \"{provider_cls_name}\"") from exception
|
||||
raise SchemaError(
|
||||
f'Can not import provider "{provider_cls_name}"'
|
||||
) from exception
|
||||
except AttributeError:
|
||||
return None
|
||||
else:
|
||||
if isinstance(cls, type) and not issubclass(cls, providers.Provider):
|
||||
raise SchemaError(f"Provider class \"{cls}\" is not a subclass of providers base class")
|
||||
raise SchemaError(
|
||||
f'Provider class "{cls}" is not a subclass of providers base class'
|
||||
)
|
||||
return cls
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -27,8 +27,9 @@ from typing import (
|
|||
if sys.version_info < (3, 7):
|
||||
from typing import GenericMeta
|
||||
else:
|
||||
class GenericMeta(type):
|
||||
...
|
||||
|
||||
class GenericMeta(type): ...
|
||||
|
||||
|
||||
# Hotfix, see: https://github.com/ets-labs/python-dependency-injector/issues/362
|
||||
if sys.version_info >= (3, 9):
|
||||
|
|
@ -99,7 +100,9 @@ class PatchedRegistry:
|
|||
def register_callable(self, patched: "PatchedCallable") -> None:
|
||||
self._callables[patched.patched] = patched
|
||||
|
||||
def get_callables_from_module(self, module: ModuleType) -> Iterator[Callable[..., Any]]:
|
||||
def get_callables_from_module(
|
||||
self, module: ModuleType
|
||||
) -> Iterator[Callable[..., Any]]:
|
||||
for patched_callable in self._callables.values():
|
||||
if not patched_callable.is_in_module(module):
|
||||
continue
|
||||
|
|
@ -114,7 +117,9 @@ class PatchedRegistry:
|
|||
def register_attribute(self, patched: "PatchedAttribute") -> None:
|
||||
self._attributes.add(patched)
|
||||
|
||||
def get_attributes_from_module(self, module: ModuleType) -> Iterator["PatchedAttribute"]:
|
||||
def get_attributes_from_module(
|
||||
self, module: ModuleType
|
||||
) -> Iterator["PatchedAttribute"]:
|
||||
for attribute in self._attributes:
|
||||
if not attribute.is_in_module(module):
|
||||
continue
|
||||
|
|
@ -139,11 +144,11 @@ class PatchedCallable:
|
|||
)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
patched: Optional[Callable[..., Any]] = None,
|
||||
original: Optional[Callable[..., Any]] = None,
|
||||
reference_injections: Optional[Dict[Any, Any]] = None,
|
||||
reference_closing: Optional[Dict[Any, Any]] = None,
|
||||
self,
|
||||
patched: Optional[Callable[..., Any]] = None,
|
||||
original: Optional[Callable[..., Any]] = None,
|
||||
reference_injections: Optional[Dict[Any, Any]] = None,
|
||||
reference_closing: Optional[Dict[Any, Any]] = None,
|
||||
) -> None:
|
||||
self.patched = patched
|
||||
self.original = original
|
||||
|
|
@ -214,18 +219,21 @@ class ProvidersMap:
|
|||
)
|
||||
|
||||
def resolve_provider(
|
||||
self,
|
||||
provider: Union[providers.Provider, str],
|
||||
modifier: Optional["Modifier"] = None,
|
||||
self,
|
||||
provider: Union[providers.Provider, str],
|
||||
modifier: Optional["Modifier"] = None,
|
||||
) -> Optional[providers.Provider]:
|
||||
if isinstance(provider, providers.Delegate):
|
||||
return self._resolve_delegate(provider)
|
||||
elif isinstance(provider, (
|
||||
providers.ProvidedInstance,
|
||||
providers.AttributeGetter,
|
||||
providers.ItemGetter,
|
||||
providers.MethodCaller,
|
||||
)):
|
||||
elif isinstance(
|
||||
provider,
|
||||
(
|
||||
providers.ProvidedInstance,
|
||||
providers.AttributeGetter,
|
||||
providers.ItemGetter,
|
||||
providers.MethodCaller,
|
||||
),
|
||||
):
|
||||
return self._resolve_provided_instance(provider)
|
||||
elif isinstance(provider, providers.ConfigurationOption):
|
||||
return self._resolve_config_option(provider)
|
||||
|
|
@ -237,9 +245,9 @@ class ProvidersMap:
|
|||
return self._resolve_provider(provider)
|
||||
|
||||
def _resolve_string_id(
|
||||
self,
|
||||
id: str,
|
||||
modifier: Optional["Modifier"] = None,
|
||||
self,
|
||||
id: str,
|
||||
modifier: Optional["Modifier"] = None,
|
||||
) -> Optional[providers.Provider]:
|
||||
if id == self.CONTAINER_STRING_ID:
|
||||
return self._container.__self__
|
||||
|
|
@ -256,16 +264,19 @@ class ProvidersMap:
|
|||
return provider
|
||||
|
||||
def _resolve_provided_instance(
|
||||
self,
|
||||
original: providers.Provider,
|
||||
self,
|
||||
original: providers.Provider,
|
||||
) -> Optional[providers.Provider]:
|
||||
modifiers = []
|
||||
while isinstance(original, (
|
||||
while isinstance(
|
||||
original,
|
||||
(
|
||||
providers.ProvidedInstance,
|
||||
providers.AttributeGetter,
|
||||
providers.ItemGetter,
|
||||
providers.MethodCaller,
|
||||
)):
|
||||
),
|
||||
):
|
||||
modifiers.insert(0, original)
|
||||
original = original.provides
|
||||
|
||||
|
|
@ -289,8 +300,8 @@ class ProvidersMap:
|
|||
return new
|
||||
|
||||
def _resolve_delegate(
|
||||
self,
|
||||
original: providers.Delegate,
|
||||
self,
|
||||
original: providers.Delegate,
|
||||
) -> Optional[providers.Provider]:
|
||||
provider = self._resolve_provider(original.provides)
|
||||
if provider:
|
||||
|
|
@ -298,9 +309,9 @@ class ProvidersMap:
|
|||
return provider
|
||||
|
||||
def _resolve_config_option(
|
||||
self,
|
||||
original: providers.ConfigurationOption,
|
||||
as_: Any = None,
|
||||
self,
|
||||
original: providers.ConfigurationOption,
|
||||
as_: Any = None,
|
||||
) -> Optional[providers.Provider]:
|
||||
original_root = original.root
|
||||
new = self._resolve_provider(original_root)
|
||||
|
|
@ -324,8 +335,8 @@ class ProvidersMap:
|
|||
return new
|
||||
|
||||
def _resolve_provider(
|
||||
self,
|
||||
original: providers.Provider,
|
||||
self,
|
||||
original: providers.Provider,
|
||||
) -> Optional[providers.Provider]:
|
||||
try:
|
||||
return self._map[original]
|
||||
|
|
@ -334,9 +345,9 @@ class ProvidersMap:
|
|||
|
||||
@classmethod
|
||||
def _create_providers_map(
|
||||
cls,
|
||||
current_container: Container,
|
||||
original_container: Container,
|
||||
cls,
|
||||
current_container: Container,
|
||||
original_container: Container,
|
||||
) -> Dict[providers.Provider, providers.Provider]:
|
||||
current_providers = current_container.providers
|
||||
current_providers["__self__"] = current_container.__self__
|
||||
|
|
@ -349,8 +360,9 @@ class ProvidersMap:
|
|||
original_provider = original_providers[provider_name]
|
||||
providers_map[original_provider] = current_provider
|
||||
|
||||
if isinstance(current_provider, providers.Container) \
|
||||
and isinstance(original_provider, providers.Container):
|
||||
if isinstance(current_provider, providers.Container) and isinstance(
|
||||
original_provider, providers.Container
|
||||
):
|
||||
subcontainer_map = cls._create_providers_map(
|
||||
current_container=current_provider.container,
|
||||
original_container=original_provider.container,
|
||||
|
|
@ -376,19 +388,21 @@ class InspectFilter:
|
|||
return werkzeug and isinstance(instance, werkzeug.local.LocalProxy)
|
||||
|
||||
def _is_starlette_request_cls(self, instance: object) -> bool:
|
||||
return starlette \
|
||||
and isinstance(instance, type) \
|
||||
and _safe_is_subclass(instance, starlette.requests.Request)
|
||||
return (
|
||||
starlette
|
||||
and isinstance(instance, type)
|
||||
and _safe_is_subclass(instance, starlette.requests.Request)
|
||||
)
|
||||
|
||||
def _is_builtin(self, instance: object) -> bool:
|
||||
return inspect.isbuiltin(instance)
|
||||
|
||||
|
||||
def wire( # noqa: C901
|
||||
container: Container,
|
||||
*,
|
||||
modules: Optional[Iterable[ModuleType]] = None,
|
||||
packages: Optional[Iterable[ModuleType]] = None,
|
||||
container: Container,
|
||||
*,
|
||||
modules: Optional[Iterable[ModuleType]] = None,
|
||||
packages: Optional[Iterable[ModuleType]] = None,
|
||||
) -> None:
|
||||
"""Wire container providers with provided packages and modules."""
|
||||
modules = [*modules] if modules else []
|
||||
|
|
@ -418,18 +432,22 @@ def wire( # noqa: C901
|
|||
else:
|
||||
for cls_member_name, cls_member in cls_members:
|
||||
if _is_marker(cls_member):
|
||||
_patch_attribute(cls, cls_member_name, cls_member, providers_map)
|
||||
_patch_attribute(
|
||||
cls, cls_member_name, cls_member, providers_map
|
||||
)
|
||||
elif _is_method(cls_member):
|
||||
_patch_method(cls, cls_member_name, cls_member, providers_map)
|
||||
_patch_method(
|
||||
cls, cls_member_name, cls_member, providers_map
|
||||
)
|
||||
|
||||
for patched in _patched_registry.get_callables_from_module(module):
|
||||
_bind_injections(patched, providers_map)
|
||||
|
||||
|
||||
def unwire( # noqa: C901
|
||||
*,
|
||||
modules: Optional[Iterable[ModuleType]] = None,
|
||||
packages: Optional[Iterable[ModuleType]] = None,
|
||||
*,
|
||||
modules: Optional[Iterable[ModuleType]] = None,
|
||||
packages: Optional[Iterable[ModuleType]] = None,
|
||||
) -> None:
|
||||
"""Wire provided packages and modules with previous wired providers."""
|
||||
modules = [*modules] if modules else []
|
||||
|
|
@ -443,7 +461,9 @@ def unwire( # noqa: C901
|
|||
if inspect.isfunction(member):
|
||||
_unpatch(module, name, member)
|
||||
elif inspect.isclass(member):
|
||||
for method_name, method in inspect.getmembers(member, inspect.isfunction):
|
||||
for method_name, method in inspect.getmembers(
|
||||
member, inspect.isfunction
|
||||
):
|
||||
_unpatch(member, method_name, method)
|
||||
|
||||
for patched in _patched_registry.get_callables_from_module(module):
|
||||
|
|
@ -462,10 +482,10 @@ def inject(fn: F) -> F:
|
|||
|
||||
|
||||
def _patch_fn(
|
||||
module: ModuleType,
|
||||
name: str,
|
||||
fn: Callable[..., Any],
|
||||
providers_map: ProvidersMap,
|
||||
module: ModuleType,
|
||||
name: str,
|
||||
fn: Callable[..., Any],
|
||||
providers_map: ProvidersMap,
|
||||
) -> None:
|
||||
if not _is_patched(fn):
|
||||
reference_injections, reference_closing = _fetch_reference_injections(fn)
|
||||
|
|
@ -479,14 +499,16 @@ def _patch_fn(
|
|||
|
||||
|
||||
def _patch_method(
|
||||
cls: Type,
|
||||
name: str,
|
||||
method: Callable[..., Any],
|
||||
providers_map: ProvidersMap,
|
||||
cls: Type,
|
||||
name: str,
|
||||
method: Callable[..., Any],
|
||||
providers_map: ProvidersMap,
|
||||
) -> None:
|
||||
if hasattr(cls, "__dict__") \
|
||||
and name in cls.__dict__ \
|
||||
and isinstance(cls.__dict__[name], (classmethod, staticmethod)):
|
||||
if (
|
||||
hasattr(cls, "__dict__")
|
||||
and name in cls.__dict__
|
||||
and isinstance(cls.__dict__[name], (classmethod, staticmethod))
|
||||
):
|
||||
method = cls.__dict__[name]
|
||||
fn = method.__func__
|
||||
else:
|
||||
|
|
@ -507,13 +529,15 @@ def _patch_method(
|
|||
|
||||
|
||||
def _unpatch(
|
||||
module: ModuleType,
|
||||
name: str,
|
||||
fn: Callable[..., Any],
|
||||
module: ModuleType,
|
||||
name: str,
|
||||
fn: Callable[..., Any],
|
||||
) -> None:
|
||||
if hasattr(module, "__dict__") \
|
||||
and name in module.__dict__ \
|
||||
and isinstance(module.__dict__[name], (classmethod, staticmethod)):
|
||||
if (
|
||||
hasattr(module, "__dict__")
|
||||
and name in module.__dict__
|
||||
and isinstance(module.__dict__[name], (classmethod, staticmethod))
|
||||
):
|
||||
method = module.__dict__[name]
|
||||
fn = method.__func__
|
||||
|
||||
|
|
@ -524,10 +548,10 @@ def _unpatch(
|
|||
|
||||
|
||||
def _patch_attribute(
|
||||
member: Any,
|
||||
name: str,
|
||||
marker: "_Marker",
|
||||
providers_map: ProvidersMap,
|
||||
member: Any,
|
||||
name: str,
|
||||
marker: "_Marker",
|
||||
providers_map: ProvidersMap,
|
||||
) -> None:
|
||||
provider = providers_map.resolve_provider(marker.provider, marker.modifier)
|
||||
if provider is None:
|
||||
|
|
@ -549,15 +573,14 @@ def _unpatch_attribute(patched: PatchedAttribute) -> None:
|
|||
|
||||
|
||||
def _fetch_reference_injections( # noqa: C901
|
||||
fn: Callable[..., Any],
|
||||
fn: Callable[..., Any],
|
||||
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
|
||||
# Hotfix, see:
|
||||
# - https://github.com/ets-labs/python-dependency-injector/issues/362
|
||||
# - https://github.com/ets-labs/python-dependency-injector/issues/398
|
||||
if GenericAlias and any((
|
||||
fn is GenericAlias,
|
||||
getattr(fn, "__func__", None) is GenericAlias
|
||||
)):
|
||||
if GenericAlias and any(
|
||||
(fn is GenericAlias, getattr(fn, "__func__", None) is GenericAlias)
|
||||
):
|
||||
fn = fn.__init__
|
||||
|
||||
try:
|
||||
|
|
@ -573,8 +596,9 @@ def _fetch_reference_injections( # noqa: C901
|
|||
injections = {}
|
||||
closing = {}
|
||||
for parameter_name, parameter in signature.parameters.items():
|
||||
if not isinstance(parameter.default, _Marker) \
|
||||
and not _is_fastapi_depends(parameter.default):
|
||||
if not isinstance(parameter.default, _Marker) and not _is_fastapi_depends(
|
||||
parameter.default
|
||||
):
|
||||
continue
|
||||
|
||||
marker = parameter.default
|
||||
|
|
@ -593,7 +617,9 @@ def _fetch_reference_injections( # noqa: C901
|
|||
return injections, closing
|
||||
|
||||
|
||||
def _locate_dependent_closing_args(provider: providers.Provider) -> Dict[str, providers.Provider]:
|
||||
def _locate_dependent_closing_args(
|
||||
provider: providers.Provider,
|
||||
) -> Dict[str, providers.Provider]:
|
||||
if not hasattr(provider, "args"):
|
||||
return {}
|
||||
|
||||
|
|
@ -647,8 +673,8 @@ def _fetch_modules(package):
|
|||
if not hasattr(package, "__path__") or not hasattr(package, "__name__"):
|
||||
return modules
|
||||
for module_info in pkgutil.walk_packages(
|
||||
path=package.__path__,
|
||||
prefix=package.__name__ + ".",
|
||||
path=package.__path__,
|
||||
prefix=package.__name__ + ".",
|
||||
):
|
||||
module = importlib.import_module(module_info.name)
|
||||
modules.append(module)
|
||||
|
|
@ -664,9 +690,9 @@ def _is_marker(member) -> bool:
|
|||
|
||||
|
||||
def _get_patched(
|
||||
fn: F,
|
||||
reference_injections: Dict[Any, Any],
|
||||
reference_closing: Dict[Any, Any],
|
||||
fn: F,
|
||||
reference_injections: Dict[Any, Any],
|
||||
reference_closing: Dict[Any, Any],
|
||||
) -> F:
|
||||
patched_object = PatchedCallable(
|
||||
original=fn,
|
||||
|
|
@ -694,9 +720,11 @@ def _is_patched(fn) -> bool:
|
|||
|
||||
|
||||
def _is_declarative_container(instance: Any) -> bool:
|
||||
return (isinstance(instance, type)
|
||||
and getattr(instance, "__IS_CONTAINER__", False) is True
|
||||
and getattr(instance, "declarative_parent", None) is None)
|
||||
return (
|
||||
isinstance(instance, type)
|
||||
and getattr(instance, "__IS_CONTAINER__", False) is True
|
||||
and getattr(instance, "declarative_parent", None) is None
|
||||
)
|
||||
|
||||
|
||||
def _safe_is_subclass(instance: Any, cls: Type) -> bool:
|
||||
|
|
@ -709,11 +737,10 @@ def _safe_is_subclass(instance: Any, cls: Type) -> bool:
|
|||
class Modifier:
|
||||
|
||||
def modify(
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
) -> providers.Provider:
|
||||
...
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
) -> providers.Provider: ...
|
||||
|
||||
|
||||
class TypeModifier(Modifier):
|
||||
|
|
@ -722,9 +749,9 @@ class TypeModifier(Modifier):
|
|||
self.type_ = type_
|
||||
|
||||
def modify(
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
) -> providers.Provider:
|
||||
return provider.as_(self.type_)
|
||||
|
||||
|
|
@ -762,9 +789,9 @@ class RequiredModifier(Modifier):
|
|||
return self
|
||||
|
||||
def modify(
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
) -> providers.Provider:
|
||||
provider = provider.required()
|
||||
if self.type_modifier:
|
||||
|
|
@ -783,9 +810,9 @@ class InvariantModifier(Modifier):
|
|||
self.id = id
|
||||
|
||||
def modify(
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
self,
|
||||
provider: providers.ConfigurationOption,
|
||||
providers_map: ProvidersMap,
|
||||
) -> providers.Provider:
|
||||
invariant_segment = providers_map.resolve_provider(self.id)
|
||||
return provider[invariant_segment]
|
||||
|
|
@ -818,9 +845,9 @@ class ProvidedInstance(Modifier):
|
|||
return self
|
||||
|
||||
def modify(
|
||||
self,
|
||||
provider: providers.Provider,
|
||||
providers_map: ProvidersMap,
|
||||
self,
|
||||
provider: providers.Provider,
|
||||
providers_map: ProvidersMap,
|
||||
) -> providers.Provider:
|
||||
provider = provider.provided
|
||||
for type_, value in self.segments:
|
||||
|
|
@ -851,9 +878,9 @@ class _Marker(Generic[T], metaclass=ClassGetItemMeta):
|
|||
__IS_MARKER__ = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
provider: Union[providers.Provider, Container, str],
|
||||
modifier: Optional[Modifier] = None,
|
||||
self,
|
||||
provider: Union[providers.Provider, Container, str],
|
||||
modifier: Optional[Modifier] = None,
|
||||
) -> None:
|
||||
if _is_declarative_container(provider):
|
||||
provider = provider.__self__
|
||||
|
|
@ -869,16 +896,13 @@ class _Marker(Generic[T], metaclass=ClassGetItemMeta):
|
|||
return self
|
||||
|
||||
|
||||
class Provide(_Marker):
|
||||
...
|
||||
class Provide(_Marker): ...
|
||||
|
||||
|
||||
class Provider(_Marker):
|
||||
...
|
||||
class Provider(_Marker): ...
|
||||
|
||||
|
||||
class Closing(_Marker):
|
||||
...
|
||||
class Closing(_Marker): ...
|
||||
|
||||
|
||||
class AutoLoader:
|
||||
|
|
@ -928,8 +952,7 @@ class AutoLoader:
|
|||
super().exec_module(module)
|
||||
loader.wire_module(module)
|
||||
|
||||
class ExtensionFileLoader(importlib.machinery.ExtensionFileLoader):
|
||||
...
|
||||
class ExtensionFileLoader(importlib.machinery.ExtensionFileLoader): ...
|
||||
|
||||
loader_details = [
|
||||
(SourcelessFileLoader, importlib.machinery.BYTECODE_SUFFIXES),
|
||||
|
|
@ -998,4 +1021,5 @@ def _get_async_patched(fn: F, patched: PatchedCallable) -> F:
|
|||
patched.injections,
|
||||
patched.closing,
|
||||
)
|
||||
|
||||
return _patched
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user