Add single container prototype

This commit is contained in:
Roman Mogylatov 2021-02-22 15:58:09 -05:00
parent 8cad8c6b65
commit 436bb1a5a2
4 changed files with 2663 additions and 1902 deletions

File diff suppressed because it is too large Load Diff

View File

@ -1,3 +1,4 @@
from pathlib import Path
from typing import (
Generic,
Type,
@ -34,6 +35,7 @@ class Container:
def __init__(self) -> None: ...
def __deepcopy__(self, memo: Optional[Dict[str, Any]]) -> Provider: ...
def __setattr__(self, name: str, value: Union[Provider, Any]) -> None: ...
def __getattr__(self, name: str) -> Provider: ...
def __delattr__(self, name: str) -> None: ...
def set_providers(self, **providers: Provider): ...
def set_provider(self, name: str, provider: Provider) -> None: ...
@ -48,6 +50,9 @@ class Container:
def apply_container_providers_overridings(self) -> None: ...
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]) -> None:
def from_json_schema(self, filepath: Union[Path, str]) -> None:
@overload
def resolve_provider_name(self, provider: Provider) -> str: ...
@classmethod

View File

@ -11,6 +11,7 @@ import six
from . import providers, errors
from .providers cimport __is_future_or_coroutine
from .schema import build_schema
if sys.version_info[:2] >= (3, 6):
@ -330,6 +331,21 @@ class DynamicContainer(Container):
f'{", ".join(undefined_names)}',
)
def from_schema(self, schema):
"""Build container providers from schema."""
for name, provider in build_schema(schema).items():
self.set_provider(name, provider)
def from_yaml_schema(self, filepath):
"""Build container providers from YAML file schema."""
# TODO
...
def from_json_schema(self, filepath):
"""Build container providers from JSON file schema."""
# TODO
...
def resolve_provider_name(self, provider):
"""Try to resolve provider name."""
for provider_name, container_provider in self.providers.items():

View File

@ -0,0 +1,120 @@
"""Schema module."""
import importlib
from typing import Dict, Any, Type, Optional
from . import providers
Schema = Dict[Any, Any]
def build_schema(schema: Schema) -> Dict[str, providers.Provider]:
"""Build provider schema."""
built = {}
for provider_name, data in schema['providers'].items():
provider_type = _get_provider_cls(data['provider'])
args = []
provides = data.get('provides')
if provides:
provides = _import_string(provides)
if provides:
args.append(provides)
provider = provider_type(*args)
built[provider_name] = provider
for provider_name, data in schema['providers'].items():
provider = built[provider_name]
args = []
kwargs = {}
arg_injections = data.get('args')
if arg_injections:
for arg in arg_injections:
injection = _resolve_provider(arg, built)
if not injection:
injection = arg
args.append(injection)
if args:
provider.add_args(*args)
kwarg_injections = data.get('kwargs')
if kwarg_injections:
for name, arg in kwarg_injections.items():
injection = _resolve_provider(arg, built)
if not injection:
injection = arg
kwargs[name] = injection
if kwargs:
provider.add_kwargs(**kwargs)
# TODO: add attributes
return built
def _resolve_provider(name: str, built: Dict[Any, providers.Provider]) -> Optional[providers.Provider]:
segments = name.split('.')
try:
provider = built[segments[0]]
except KeyError:
return None
for segment in segments[1:]:
if segment == 'as_int()':
provider = provider.as_int()
elif segment == 'as_float()':
provider = provider.as_float()
elif segment.startswith('is_'): # TODO
provider = provider.as_(str)
...
else:
try:
provider = getattr(provider, segment)
except AttributeError:
return None
return provider
def _get_provider_cls(provider_cls_name: str) -> Type[providers.Provider]:
std_provider_type = _fetch_provider_cls_from_std(provider_cls_name)
if std_provider_type:
return std_provider_type
custom_provider_type = _import_provider_cls(provider_cls_name)
if custom_provider_type:
return custom_provider_type
raise SchemaError(f'Undefined provider class: "{provider_cls_name}"')
def _fetch_provider_cls_from_std(provider_cls_name: str) -> Optional[Type[providers.Provider]]:
return getattr(providers, provider_cls_name, None)
def _import_provider_cls(provider_cls_name: str) -> Optional[Type[providers.Provider]]:
try:
cls = _import_string(provider_cls_name)
except (ImportError, ValueError) as 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')
return cls
def _import_string(string_name: str) -> Optional[object]:
segments = string_name.split('.')
module_name = '.'.join(segments[:-1])
member = segments[-1]
module = importlib.import_module(module_name)
return getattr(module, member, None)
class SchemaError(Exception):
"""Schema-related error."""