schema.py

"""Schema module."""

import builtins
import importlib
from typing import Dict, Any, Type, Optional

from . import containers, providers


ContainerSchema = Dict[Any, Any]
ProviderSchema = Dict[Any, Any]


class SchemaProcessorV1:

    def __init__(self, schema: ContainerSchema) -> None:
        self._schema = schema
        self._container = containers.DynamicContainer()

    def process(self):
        """Process schema."""
        self._create_providers(self._schema["container"])
        self._setup_injections(self._schema["container"])

    def get_providers(self):
        """Return providers."""
        return self._container.providers

    def _create_providers(
        self,
        provider_schema: ProviderSchema,
        container: Optional[containers.Container] = None,
    ) -> None:
        if container is None:
            container = self._container
        for provider_name, data in provider_schema.items():
            provider = None

            if "provider" in data:
                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)

            if provider is None:
                provider = providers.Container(containers.DynamicContainer)

            container.set_provider(provider_name, provider)

            if isinstance(provider, providers.Container):
                self._create_providers(provider_schema=data, container=provider)

    def _setup_injections(  # noqa: C901
        self,
        provider_schema: ProviderSchema,
        container: Optional[containers.Container] = None,
    ) -> None:
        if container is None:
            container = self._container

        for provider_name, data in provider_schema.items():
            provider = getattr(container, provider_name)
            args = []
            kwargs = {}

            provides = data.get("provides")
            if provides:
                if isinstance(provides, str) and provides.startswith("container."):
                    provides = self._resolve_provider(provides[len("container.") :])
                else:
                    provides = _import_string(provides)
                provider.set_provides(provides)

            arg_injections = data.get("args")
            if arg_injections:
                for arg in arg_injections:
                    injection = None

                    if isinstance(arg, str) and arg.startswith("container."):
                        injection = self._resolve_provider(arg[len("container.") :])

                    # TODO: refactoring
                    if isinstance(arg, dict):
                        provider_args = []
                        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.") :]
                                )
                            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."):
                                provider_args.append(
                                    self._resolve_provider(
                                        provider_arg[len("container.") :]
                                    ),
                                )
                        injection = provider_type(*provider_args)

                    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 = None

                    if isinstance(arg, str) and arg.startswith("container."):
                        injection = self._resolve_provider(arg[len("container.") :])

                    # TODO: refactoring
                    if isinstance(arg, dict):
                        provider_args = []
                        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.") :]
                                )
                            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."):
                                provider_args.append(
                                    self._resolve_provider(
                                        provider_arg[len("container.") :]
                                    ),
                                )
                        injection = provider_type(*provider_args)

                    if not injection:
                        injection = arg

                    kwargs[name] = injection
            if kwargs:
                provider.add_kwargs(**kwargs)

            if isinstance(provider, providers.Container):
                self._setup_injections(provider_schema=data, container=provider)

    def _resolve_provider(self, name: str) -> Optional[providers.Provider]:
        segments = name.split(".")
        try:
            provider = getattr(self._container, segments[0])
        except AttributeError:
            return None

        for segment in segments[1:]:
            parentheses = ""
            if "(" in segment and ")" in segment:
                parentheses = segment[segment.find("(") : segment.rfind(")") + 1]
                segment = segment.replace(parentheses, "")

            try:
                provider = getattr(provider, segment)
            except AttributeError:
                # TODO
                return None

            if parentheses:
                # TODO
                provider = provider()

        return provider


def build_schema(schema: ContainerSchema) -> Dict[str, providers.Provider]:
    """Build provider schema."""
    schema_processor = SchemaProcessorV1(schema)
    schema_processor.process()
    return schema_processor.get_providers()


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(".")

    if len(segments) == 1:
        member = getattr(builtins, segments[0], None)
        if member:
            return member

    module_name = ".".join(segments[:-1])
    if not module_name:
        return None

    member = segments[-1]
    module = importlib.import_module(module_name)
    return getattr(module, member, None)


class SchemaError(Exception):
    """Schema-related error."""