Add multiple containers prototype

This commit is contained in:
Roman Mogylatov 2021-02-22 17:03:45 -05:00
parent 436bb1a5a2
commit 5121b5c070
3 changed files with 1110 additions and 1084 deletions

File diff suppressed because it is too large Load Diff

View File

@ -11,7 +11,6 @@ import six
from . import providers, errors from . import providers, errors
from .providers cimport __is_future_or_coroutine from .providers cimport __is_future_or_coroutine
from .schema import build_schema
if sys.version_info[:2] >= (3, 6): if sys.version_info[:2] >= (3, 6):
@ -333,6 +332,7 @@ class DynamicContainer(Container):
def from_schema(self, schema): def from_schema(self, schema):
"""Build container providers from schema.""" """Build container providers from schema."""
from .schema import build_schema
for name, provider in build_schema(schema).items(): for name, provider in build_schema(schema).items():
self.set_provider(name, provider) self.set_provider(name, provider)

View File

@ -3,7 +3,7 @@
import importlib import importlib
from typing import Dict, Any, Type, Optional from typing import Dict, Any, Type, Optional
from . import providers from . import containers, providers
Schema = Dict[Any, Any] Schema = Dict[Any, Any]
@ -11,9 +11,17 @@ Schema = Dict[Any, Any]
def build_schema(schema: Schema) -> Dict[str, providers.Provider]: def build_schema(schema: Schema) -> Dict[str, providers.Provider]:
"""Build provider schema.""" """Build provider schema."""
built = {} container = containers.DynamicContainer()
_create_providers(container, schema['providers'])
_setup_injections(container, schema['providers'])
return container.providers
for provider_name, data in schema['providers'].items():
def _create_providers(
container: containers.Container,
providers_data: Dict[str, Any],
):
for provider_name, data in providers_data.items():
provider_type = _get_provider_cls(data['provider']) provider_type = _get_provider_cls(data['provider'])
args = [] args = []
@ -23,18 +31,35 @@ def build_schema(schema: Schema) -> Dict[str, providers.Provider]:
if provides: if provides:
args.append(provides) args.append(provides)
provider = provider_type(*args) if provider_type is providers.Container:
built[provider_name] = provider provides = containers.DynamicContainer
args.append(provides)
for provider_name, data in schema['providers'].items(): provider = provider_type(*args)
provider = built[provider_name] container.set_provider(provider_name, provider)
if isinstance(provider, providers.Container):
_create_providers(provider, data['providers'])
def _setup_injections(
container: containers.Container,
providers_data: Dict[str, Any],
*,
current_container: Optional[containers.Container] = None,
):
if not current_container:
current_container = container
for provider_name, data in providers_data.items():
provider = getattr(current_container, provider_name)
args = [] args = []
kwargs = {} kwargs = {}
arg_injections = data.get('args') arg_injections = data.get('args')
if arg_injections: if arg_injections:
for arg in arg_injections: for arg in arg_injections:
injection = _resolve_provider(arg, built) injection = _resolve_provider(container, arg)
if not injection: if not injection:
injection = arg injection = arg
args.append(injection) args.append(injection)
@ -44,23 +69,22 @@ def build_schema(schema: Schema) -> Dict[str, providers.Provider]:
kwarg_injections = data.get('kwargs') kwarg_injections = data.get('kwargs')
if kwarg_injections: if kwarg_injections:
for name, arg in kwarg_injections.items(): for name, arg in kwarg_injections.items():
injection = _resolve_provider(arg, built) injection = _resolve_provider(container, arg)
if not injection: if not injection:
injection = arg injection = arg
kwargs[name] = injection kwargs[name] = injection
if kwargs: if kwargs:
provider.add_kwargs(**kwargs) provider.add_kwargs(**kwargs)
# TODO: add attributes if isinstance(provider, providers.Container):
_setup_injections(container, data['providers'], current_container=provider)
return built
def _resolve_provider(name: str, built: Dict[Any, providers.Provider]) -> Optional[providers.Provider]: def _resolve_provider(container: containers.Container, name: str) -> Optional[providers.Provider]:
segments = name.split('.') segments = name.split('.')
try: try:
provider = built[segments[0]] provider = getattr(container, segments[0])
except KeyError: except AttributeError:
return None return None
for segment in segments[1:]: for segment in segments[1:]: