9377842010

Committed 05 Jun 2024 03:41AM UTC coverage: 0.0% (-92.0%) from 92.016%

Build # 9377842010

Build Type

Pull #765

github

Committed by

rmk135

Commit Message

Remove pypy 3.9

Pull Request Pull Request #765: Add Python 3.12 Support (#752)

Run Details

0 of 764 relevant lines covered (0.0%)

0.0 hits per line

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

0.0

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

ets-labs / python-dependency-injector / 9377842010

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