3726562862

Build Type

push

github-actions

Committed by Roman Mogylatov

Commit Message

Update publishing job

Run Details

691 of 750 relevant lines covered (92.13%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.39

/src/dependency_injector/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."""

1	"""Schema module."""
2
3	import builtins	1✔
4	import importlib	1✔
5	from typing import Dict, Any, Type, Optional	1✔
6
7	from . import containers, providers	1✔
8
9
10	ContainerSchema = Dict[Any, Any]	1✔
11	ProviderSchema = Dict[Any, Any]	1✔
12
13
14	class SchemaProcessorV1:	1✔
15
16	def __init__(self, schema: ContainerSchema) -> None:	1✔
17	self._schema = schema	1✔
18	self._container = containers.DynamicContainer()	1✔
19
20	def process(self):	1✔
21	"""Process schema."""
22	self._create_providers(self._schema["container"])	1✔
23	self._setup_injections(self._schema["container"])	1✔
24
25	def get_providers(self):	1✔
26	"""Return providers."""
27	return self._container.providers	1✔
28
29	def _create_providers(	1✔
30	self,
31	provider_schema: ProviderSchema,
32	container: Optional[containers.Container] = None,
33	) -> None:
34	if container is None:	1✔
35	container = self._container	1✔
36	for provider_name, data in provider_schema.items():	1✔
37	provider = None	1✔
38
39	if "provider" in data:	1✔
40	provider_type = _get_provider_cls(data["provider"])	1✔
41	args = []	1✔
42
43	# provides = data.get("provides")
44	# if provides:
45	# provides = _import_string(provides)
46	# if provides:
47	# args.append(provides)
48
49	provider = provider_type(*args)	1✔
50
51	if provider is None:	1✔
52	provider = providers.Container(containers.DynamicContainer)	1✔
53
54	container.set_provider(provider_name, provider)	1✔
55
56	if isinstance(provider, providers.Container):	1✔
57	self._create_providers(provider_schema=data, container=provider)	1✔
58
59	def _setup_injections( # noqa: C901	1✔
60	self,
61	provider_schema: ProviderSchema,
62	container: Optional[containers.Container] = None,
63	) -> None:
64	if container is None:	1✔
65	container = self._container	1✔
66
67	for provider_name, data in provider_schema.items():	1✔
68	provider = getattr(container, provider_name)	1✔
69	args = []	1✔
70	kwargs = {}	1✔
71
72	provides = data.get("provides")	1✔
73	if provides:	1✔
74	if isinstance(provides, str) and provides.startswith("container."):	1✔
75	provides = self._resolve_provider(provides[len("container."):])	1✔
76	else:
77	provides = _import_string(provides)	1✔
78	provider.set_provides(provides)	1✔
79
80	arg_injections = data.get("args")	1✔
81	if arg_injections:	1✔
82	for arg in arg_injections:	1✔
83	injection = None	1✔
84
85	if isinstance(arg, str) and arg.startswith("container."):	1✔
86	injection = self._resolve_provider(arg[len("container."):])	1✔
87
88	# TODO: refactoring
89	if isinstance(arg, dict):	1✔
90	provider_args = []	1✔
91	provider_type = _get_provider_cls(arg.get("provider"))	1✔
92	provides = arg.get("provides")	1✔
93	if provides:	1✔
94	if isinstance(provides, str) and provides.startswith("container."):	1✔
95	provides = self._resolve_provider(provides[len("container."):])	×
96	else:
97	provides = _import_string(provides)	1✔
98	provider_args.append(provides)	1✔
99	for provider_arg in arg.get("args", []):	1✔
100	if isinstance(provider_arg, str) \	1✔
101	and provider_arg.startswith("container."):
102	provider_args.append(	1✔
103	self._resolve_provider(provider_arg[len("container."):]),
104	)
105	injection = provider_type(*provider_args)	1✔
106
107	if not injection:	1✔
108	injection = arg	1✔
109
110	args.append(injection)	1✔
111	if args:	1✔
112	provider.add_args(*args)	1✔
113
114	kwarg_injections = data.get("kwargs")	1✔
115	if kwarg_injections:	1✔
116	for name, arg in kwarg_injections.items():	1✔
117	injection = None	1✔
118
119	if isinstance(arg, str) and arg.startswith("container."):	1✔
120	injection = self._resolve_provider(arg[len("container."):])	1✔
121
122	# TODO: refactoring
123	if isinstance(arg, dict):	1✔
124	provider_args = []	1✔
125	provider_type = _get_provider_cls(arg.get("provider"))	1✔
126	provides = arg.get("provides")	1✔
127	if provides:	1✔
128	if isinstance(provides, str) and provides.startswith("container."):	1✔
129	provides = self._resolve_provider(provides[len("container."):])	×
130	else:
131	provides = _import_string(provides)	1✔
132	provider_args.append(provides)	1✔
133	for provider_arg in arg.get("args", []):	1✔
134	if isinstance(provider_arg, str) \	1✔
135	and provider_arg.startswith("container."):
136	provider_args.append(	1✔
137	self._resolve_provider(provider_arg[len("container."):]),
138	)
139	injection = provider_type(*provider_args)	1✔
140
141	if not injection:	1✔
142	injection = arg	1✔
143
144	kwargs[name] = injection	1✔
145	if kwargs:	1✔
146	provider.add_kwargs(**kwargs)	1✔
147
148	if isinstance(provider, providers.Container):	1✔
149	self._setup_injections(provider_schema=data, container=provider)	1✔
150
151	def _resolve_provider(self, name: str) -> Optional[providers.Provider]:	1✔
152	segments = name.split(".")	1✔
153	try:	1✔
154	provider = getattr(self._container, segments[0])	1✔
155	except AttributeError:	×
156	return None	×
157
158	for segment in segments[1:]:	1✔
159	parentheses = ""	1✔
160	if "(" in segment and ")" in segment:	1✔
161	parentheses = segment[segment.find("("):segment.rfind(")")+1]	1✔
162	segment = segment.replace(parentheses, "")	1✔
163
164	try:	1✔
165	provider = getattr(provider, segment)	1✔
166	except AttributeError:	×
167	# TODO
168	return None	×
169
170	if parentheses:	1✔
171	# TODO
172	provider = provider()	1✔
173
174	return provider	1✔
175
176
177	def build_schema(schema: ContainerSchema) -> Dict[str, providers.Provider]:	1✔
178	"""Build provider schema."""
179	schema_processor = SchemaProcessorV1(schema)	1✔
180	schema_processor.process()	1✔
181	return schema_processor.get_providers()	1✔
182
183
184	def _get_provider_cls(provider_cls_name: str) -> Type[providers.Provider]:	1✔
185	std_provider_type = _fetch_provider_cls_from_std(provider_cls_name)	1✔
186	if std_provider_type:	1✔
187	return std_provider_type	1✔
188
189	custom_provider_type = _import_provider_cls(provider_cls_name)	×
190	if custom_provider_type:	×
191	return custom_provider_type	×
192
193	raise SchemaError(f"Undefined provider class \"{provider_cls_name}\"")	×
194
195
196	def _fetch_provider_cls_from_std(provider_cls_name: str) -> Optional[Type[providers.Provider]]:	1✔
197	return getattr(providers, provider_cls_name, None)	1✔
198
199
200	def _import_provider_cls(provider_cls_name: str) -> Optional[Type[providers.Provider]]:	1✔
201	try:	×
202	cls = _import_string(provider_cls_name)	×
203	except (ImportError, ValueError) as exception:	×
204	raise SchemaError(f"Can not import provider \"{provider_cls_name}\"") from exception	×
205	except AttributeError:	×
206	return None	×
207	else:
208	if isinstance(cls, type) and not issubclass(cls, providers.Provider):	×
209	raise SchemaError(f"Provider class \"{cls}\" is not a subclass of providers base class")	×
210	return cls	×
211
212
213	def _import_string(string_name: str) -> Optional[object]:	1✔
214	segments = string_name.split(".")	1✔
215
216	if len(segments) == 1:	1✔
217	member = getattr(builtins, segments[0], None)	1✔
218	if member:	1✔
219	return member	1✔
220
221	module_name = ".".join(segments[:-1])	1✔
222	if not module_name:	1✔
223	return None	×
224
225	member = segments[-1]	1✔
226	module = importlib.import_module(module_name)	1✔
227	return getattr(module, member, None)	1✔
228
229
230	class SchemaError(Exception):	1✔
231	"""Schema-related error."""

ets-labs / python-dependency-injector / 3726562862

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous