lagom.container.EmptyDefinitionSet.get_definition() - Code Metrics - Inspection of "WIP Speed up context partials" - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#226)

by Steve

created 2023-03-08 13:02 UTC

EmptyDefinitionSet.get_definition() A

↳ Parent: lagom.container

Complexity

Conditions

Size

Total Lines	7
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	7
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

import functools
import io
import logging
import typing

from .compilaton import mypyc_attr

from types import FunctionType, MethodType
from typing import (
    Dict,
    Type,
    Any,
    TypeVar,
    Callable,
    Set,
    List,
    Optional,
    cast,
    Union,
)

from .definitions import (
    normalise,
    Singleton,
    Alias,
    ConstructionWithoutContainer,
    UnresolvableTypeDefinition,
)
from .exceptions import (
    UnresolvableType,
    DuplicateDefinition,
    InvalidDependencyDefinition,
    RecursiveDefinitionError,
    DependencyNotDefined,
    TypeOnlyAvailableAsAwaitable,
)
from .injection_context import TemporaryInjectionContext
from .interfaces import (
    SpecialDepDefinition,
    WriteableContainer,
    TypeResolver,
    DefinitionsSource,
    ExtendableContainer,
    ContainerDebugInfo,
    CallTimeContainerUpdate,
    ContainerBoundFunction,
)
from .markers import injectable
from .updaters import update_container_singletons
from .util.logging import NullLogger
from .util.reflection import (
    FunctionSpec,
    CachingReflector,
    remove_optional_type,
    remove_awaitable_type,
)
from .wrapping import apply_argument_updater

UNRESOLVABLE_TYPES = [
    str,
    int,
    float,
    bool,
    bytes,
    bytearray,
    io.BytesIO,
    io.BufferedIOBase,
    io.BufferedRandom,
    io.BufferedReader,
    io.BufferedRWPair,
    io.BufferedWriter,
    io.FileIO,
    io.IOBase,
    io.RawIOBase,
    io.TextIOBase,
    typing.IO,
    typing.TextIO,
    typing.BinaryIO,
]

X = TypeVar("X")

Unset: Any = object()


@mypyc_attr(allow_interpreted_subclasses=True)
class Container(
    WriteableContainer, ExtendableContainer, DefinitionsSource, ContainerDebugInfo
):
    """Dependency injection container

    Lagom is a dependency injection container designed to give you "just enough"
    help with building your dependencies. The intention is that almost
    all of your code doesn't know about or rely on lagom. Lagom will
    only be involved at the top level to pull everything together.

    >>> from tests.examples import SomeClass
    >>> c = Container()
    >>> c[SomeClass]
    <tests.examples.SomeClass object at ...>

    Objects are constructed as they are needed

    >>> from tests.examples import SomeClass
    >>> c = Container()
    >>> first = c[SomeClass]
    >>> second = c[SomeClass]
    >>> first != second
    True

    And construction logic can be defined
    >>> from tests.examples import SomeClass, SomeExtendedClass
    >>> c = Container()
    >>> c[SomeClass] = SomeExtendedClass
    >>> c[SomeClass]
    <tests.examples.SomeExtendedClass object at ...>
    """

    _registered_types: Dict[Type, SpecialDepDefinition]
    _parent_definitions: DefinitionsSource
    _reflector: CachingReflector
    _undefined_logger: logging.Logger

    def __init__(
        self,
        container: Optional["Container"] = None,
        log_undefined_deps: Union[bool, logging.Logger] = False,
    ):
        """
        :param container: Optional container if provided the existing definitions will be copied
        :param log_undefined_deps indicates if a log message should be emmited when an undefined dep is loaded
        """

        # ContainerDebugInfo is always registered
        # This means consumers can consume an overview of the container
        # without hacking anything custom together.
        self._registered_types = {
            ContainerDebugInfo: ConstructionWithoutContainer(lambda: self)
        }

        if container:
            self._parent_definitions = container
            self._reflector = container._reflector
        else:
            self._parent_definitions = EmptyDefinitionSet()
            self._reflector = CachingReflector()

        if not log_undefined_deps:
            self._undefined_logger = NullLogger()
        elif log_undefined_deps is True:
            self._undefined_logger = logging.getLogger(__name__)
        else:
            self._undefined_logger = cast(logging.Logger, log_undefined_deps)

    def define(self, dep: Type[X], resolver: TypeResolver[X]) -> SpecialDepDefinition:
        """Register how to construct an object of type X

        >>> from tests.examples import SomeClass
        >>> c = Container()
        >>> c.define(SomeClass, lambda: SomeClass())
        <lagom.definitions.ConstructionWithoutContainer ...>

        :param dep: The type to be constructed
        :param resolver: A definition of how to construct it
        :return:
        """
        if dep in UNRESOLVABLE_TYPES:
            raise InvalidDependencyDefinition()
        if dep in self._registered_types:
            raise DuplicateDefinition()
        if dep is resolver:
            # This is a special case for things like container[Foo] = Foo
            return self.define(dep, Alias(dep, skip_definitions=True))
        definition = normalise(resolver)
        self._registered_types[dep] = definition
        self._registered_types[Optional[dep]] = definition  # type: ignore

        # For awaitables we add a convenience exception to be thrown if code hints on the type
        # without the awaitable.
        awaitable_type = remove_awaitable_type(dep)
        if awaitable_type:
            # Unless there's already a sync version defined.
            if awaitable_type not in self.defined_types:
                self._registered_types[awaitable_type] = UnresolvableTypeDefinition(
                    TypeOnlyAvailableAsAwaitable(awaitable_type)
                )
        return definition

    @property
    def defined_types(self) -> Set[Type]:
        """The types the container has explicit build instructions for

        :return:
        """
        return self._parent_definitions.defined_types.union(
            self._registered_types.keys()
        )

    @property
    def reflection_cache_overview(self) -> Dict[str, str]:
        return self._reflector.overview_of_cache

    def temporary_singletons(
        self, singletons: Optional[List[Type]] = None
    ) -> "TemporaryInjectionContext":
        """
        Returns a context that loads a new container with singletons that only exist
        for the context.

        >>> from tests.examples import SomeClass
        >>> base_container = Container()
        >>> def my_func():
        ...     with base_container.temporary_singletons([SomeClass]) as c:
        ...         assert c[SomeClass] is c[SomeClass]
        >>> my_func()

        :param singletons: items which should be considered singletons within the context
        :return:
        """
        updater = (
            functools.partial(update_container_singletons, singletons=singletons)
            if singletons
            else None
        )
        return TemporaryInjectionContext(self, updater)

    def resolve(
        self, dep_type: Type[X], suppress_error=False, skip_definitions=False
    ) -> X:
        """Constructs an object of type X

         If the object can't be constructed an exception will be raised unless
         supress errors is true

        >>> from tests.examples import SomeClass
        >>> c = Container()
        >>> c.resolve(SomeClass)
        <tests.examples.SomeClass object at ...>

        >>> from tests.examples import SomeClass
        >>> c = Container()
        >>> c.resolve(int)
        Traceback (most recent call last):
        ...
        lagom.exceptions.UnresolvableType: ...

        Optional wrappers are stripped out to be what is being asked for
        >>> from tests.examples import SomeClass
        >>> c = Container()
        >>> c.resolve(Optional[SomeClass])
        <tests.examples.SomeClass object at ...>

        :param dep_type: The type of object to construct
        :param suppress_error: if true returns None on failure
        :param skip_definitions:
        :return:
        """
        if not skip_definitions:
            definition = self.get_definition(dep_type)
            if definition:
                return definition.get_instance(self)

        optional_dep_type = remove_optional_type(dep_type)
        if optional_dep_type:
            return self.resolve(optional_dep_type, suppress_error=True)

        return self._reflection_build_with_err_handling(dep_type, suppress_error)

    def partial(
        self,
        func: Callable[..., X],
        shared: Optional[List[Type]] = None,
        container_updater: Optional[CallTimeContainerUpdate] = None,
    ) -> ContainerBoundFunction[X]:
        """Takes a callable and returns a callable bound to the container
        When invoking the new callable if any arguments have a default set
        to the special marker object "injectable" then they will be constructed by
        the container. For automatic injection without the marker use "magic_partial"
        >>> from tests.examples import SomeClass
        >>> c = Container()
        >>> def my_func(something: SomeClass = injectable):
        ...     return f"Successfully called with {something}"
        >>> bound_func = c.magic_partial(my_func)
        >>> bound_func()
        'Successfully called with <tests.examples.SomeClass object at ...>'

        :param func: the function to bind to the container
        :param shared: items which should be considered singletons on a per call level
        :param container_updater: An optional callable to update the container before resolution
        :return:
        """
        spec = self._get_spec_without_self(func)
        keys_to_bind = (
            key for (key, arg) in spec.defaults.items() if arg is injectable

        )
        keys_and_types = [(key, spec.annotations[key]) for key in keys_to_bind]

        base_injection_context = self.temporary_singletons(shared)
        update_container = container_updater if container_updater else _update_nothing

        def _update_args(_injection_context, supplied_args, supplied_kwargs):
            keys_to_skip = set(supplied_kwargs.keys())
            keys_to_skip.update(spec.args[0 : len(supplied_args)])
            with _injection_context as invocation_container:
                update_container(invocation_container, supplied_args, supplied_kwargs)
                kwargs = {
                    key: invocation_container.resolve(dep_type)
                    for (key, dep_type) in keys_and_types
                    if key not in keys_to_skip
                }
            kwargs.update(supplied_kwargs)
            return supplied_args, kwargs

        return apply_argument_updater(func, base_injection_context, _update_args, spec)

    def magic_partial(
        self,
        func: Callable[..., X],
        shared: Optional[List[Type]] = None,
        keys_to_skip: Optional[List[str]] = None,
        skip_pos_up_to: int = 0,
        container_updater: Optional[CallTimeContainerUpdate] = None,
    ) -> ContainerBoundFunction[X]:
        """Takes a callable and returns a callable bound to the container
        When invoking the new callable if any arguments can be constructed by the container
        then they can be ommited.
        >>> from tests.examples import SomeClass
        >>> c = Container()
        >>> def my_func(something: SomeClass):
        ...   return f"Successfully called with {something}"
        >>> bound_func = c.magic_partial(my_func)
        >>> bound_func()
        'Successfully called with <tests.examples.SomeClass object at ...>'

        :param func: the function to bind to the container
        :param shared: items which should be considered singletons on a per call level
        :param keys_to_skip: named arguments which the container shouldnt build
        :param skip_pos_up_to: positional arguments which the container shouldnt build
        :param container_updater: An optional callable to update the container before resolution
        :return:
        """
        spec = self._get_spec_without_self(func)

        update_container = container_updater if container_updater else _update_nothing
        base_injection_context = self.temporary_singletons(shared)

        def _update_args(_injection_context, supplied_args, supplied_kwargs):
            final_keys_to_skip = (keys_to_skip or []) + list(supplied_kwargs.keys())
            final_skip_pos_up_to = max(skip_pos_up_to, len(supplied_args))
            with _injection_context as invocation_container:
                update_container(invocation_container, supplied_args, supplied_kwargs)
                kwargs = invocation_container._infer_dependencies(
                    spec,
                    suppress_error=True,
                    keys_to_skip=final_keys_to_skip,
                    skip_pos_up_to=final_skip_pos_up_to,
                )
            kwargs.update(supplied_kwargs)
            return supplied_args, kwargs

        return apply_argument_updater(
            func, base_injection_context, _update_args, spec, catch_errors=True
        )

    def clone(self) -> "Container":
        """returns a copy of the container
        :return:
        """
        return Container(self, log_undefined_deps=self._undefined_logger)

    def get_definition(self, dep_type: Type[X]) -> Optional[SpecialDepDefinition[X]]:
        """
        Will return the definition in this container. If none has been defined any
        definition in the parent container will be used.

        :param dep_type:
        :return:
        """
        definition = self._registered_types.get(dep_type, Unset)
        if definition is Unset:
            return self._parent_definitions.get_definition(dep_type)
        return definition

    def __getitem__(self, dep: Type[X]) -> X:
        return self.resolve(dep)

    def __setitem__(self, dep: Type[X], resolver: TypeResolver[X]):
        self.define(dep, resolver)

    def _reflection_build_with_err_handling(
        self, dep_type: Type[X], suppress_error: bool
    ) -> X:
        try:
            if dep_type in UNRESOLVABLE_TYPES:
                raise UnresolvableType(dep_type)
            return self._reflection_build(dep_type)
        except UnresolvableType as inner_error:
            if not suppress_error:
                raise UnresolvableType(dep_type) from inner_error
            return None  # type: ignore
        except RecursionError as recursion_error:

            raise RecursiveDefinitionError(dep_type) from recursion_error

    def _reflection_build(self, dep_type: Type[X]) -> X:
        self._undefined_logger.warning(
            f"Undefined dependency. Using reflection for {dep_type}",
            extra={"undefined_dependency": dep_type},
        )
        spec = self._reflector.get_function_spec(dep_type.__init__)
        sub_deps = self._infer_dependencies(spec, types_to_skip={dep_type})
        try:
            return dep_type(**sub_deps)  # type: ignore
        except TypeError as type_error:
            raise UnresolvableType(dep_type) from type_error

    def _infer_dependencies(
        self,
        spec: FunctionSpec,
        suppress_error=False,
        keys_to_skip: Optional[List[str]] = None,
        skip_pos_up_to=0,
        types_to_skip: Optional[Set[Type]] = None,
    ):
        dep_keys_to_skip: List[str] = []
        dep_keys_to_skip.extend(spec.args[0:skip_pos_up_to])
        dep_keys_to_skip.extend(keys_to_skip or [])
        types_to_skip = types_to_skip or set()
        sub_deps = {
            key: self.resolve(sub_dep_type, suppress_error=suppress_error)
            for (key, sub_dep_type) in spec.annotations.items()
            if sub_dep_type != Any
            and (key not in dep_keys_to_skip)
            and (sub_dep_type not in types_to_skip)
        }
        return {key: dep for (key, dep) in sub_deps.items() if dep is not None}

    def _get_spec_without_self(self, func: Callable[..., X]) -> FunctionSpec:
        if isinstance(func, (FunctionType, MethodType)):
            return self._reflector.get_function_spec(func)
        t = cast(Type[X], func)
        return self._reflector.get_function_spec(t.__init__).without_argument("self")


@mypyc_attr(allow_interpreted_subclasses=True)
class ExplicitContainer(Container):
    def resolve(
        self, dep_type: Type[X], suppress_error=False, skip_definitions=False
    ) -> X:
        definition = self.get_definition(dep_type)
        if not definition:
            if suppress_error:
                return None  # type: ignore
            raise DependencyNotDefined(dep_type)
        return definition.get_instance(self)

    def define(self, dep, resolver):
        definition = super().define(dep, resolver)
        if isinstance(definition, Alias):
            raise InvalidDependencyDefinition(
                "Aliases are not valid in an explicit container"
            )
        if isinstance(definition, Singleton) and isinstance(
            definition.singleton_type, Alias
        ):
            raise InvalidDependencyDefinition(
                "Aliases are not valid inside singletons in an explicit container"
            )
        return definition

    def clone(self):
        """returns a copy of the container
        :return:
        """
        return ExplicitContainer(self, log_undefined_deps=self._undefined_logger)


class EmptyDefinitionSet(DefinitionsSource):
    """
    Represents the starting state for a collection of dependency definitions
    i.e. None and everything has to be built with reflection
    """

    def get_definition(self, dep_type: Type[X]) -> Optional[SpecialDepDefinition[X]]:
        """
        No types are defined in the empty set
        :param dep_type:
        :return:
        """
        return None

    @property
    def defined_types(self) -> Set[Type]:
        return set()


def _update_nothing(_c: WriteableContainer, _a: typing.Collection, _k: Dict):
    return None


1	1	import functools
2	1	import io
3	1	import logging
4	1	import typing
5
6	1	from .compilaton import mypyc_attr
7
8	1	from types import FunctionType, MethodType
9	1	from typing import (
10		Dict,
11		Type,
12		Any,
13		TypeVar,
14		Callable,
15		Set,
16		List,
17		Optional,
18		cast,
19		Union,
20		)
21
22	1	from .definitions import (
23		normalise,
24		Singleton,
25		Alias,
26		ConstructionWithoutContainer,
27		UnresolvableTypeDefinition,
28		)
29	1	from .exceptions import (
30		UnresolvableType,
31		DuplicateDefinition,
32		InvalidDependencyDefinition,
33		RecursiveDefinitionError,
34		DependencyNotDefined,
35		TypeOnlyAvailableAsAwaitable,
36		)
37	1	from .injection_context import TemporaryInjectionContext
38	1	from .interfaces import (
39		SpecialDepDefinition,
40		WriteableContainer,
41		TypeResolver,
42		DefinitionsSource,
43		ExtendableContainer,
44		ContainerDebugInfo,
45		CallTimeContainerUpdate,
46		ContainerBoundFunction,
47		)
48	1	from .markers import injectable
49	1	from .updaters import update_container_singletons
50	1	from .util.logging import NullLogger
51	1	from .util.reflection import (
52		FunctionSpec,
53		CachingReflector,
54		remove_optional_type,
55		remove_awaitable_type,
56		)
57	1	from .wrapping import apply_argument_updater
58
59	1	UNRESOLVABLE_TYPES = [
60		str,
61		int,
62		float,
63		bool,
64		bytes,
65		bytearray,
66		io.BytesIO,
67		io.BufferedIOBase,
68		io.BufferedRandom,
69		io.BufferedReader,
70		io.BufferedRWPair,
71		io.BufferedWriter,
72		io.FileIO,
73		io.IOBase,
74		io.RawIOBase,
75		io.TextIOBase,
76		typing.IO,
77		typing.TextIO,
78		typing.BinaryIO,
79		]
80
81	1	X = TypeVar("X")
82
83	1	Unset: Any = object()
84
85
86	1	@mypyc_attr(allow_interpreted_subclasses=True)
87	1	class Container(
88		WriteableContainer, ExtendableContainer, DefinitionsSource, ContainerDebugInfo
89		):
90		"""Dependency injection container
91
92		Lagom is a dependency injection container designed to give you "just enough"
93		help with building your dependencies. The intention is that almost
94		all of your code doesn't know about or rely on lagom. Lagom will
95		only be involved at the top level to pull everything together.
96
97		>>> from tests.examples import SomeClass
98		>>> c = Container()
99		>>> c[SomeClass]
100		<tests.examples.SomeClass object at ...>
101
102		Objects are constructed as they are needed
103
104		>>> from tests.examples import SomeClass
105		>>> c = Container()
106		>>> first = c[SomeClass]
107		>>> second = c[SomeClass]
108		>>> first != second
109		True
110
111		And construction logic can be defined
112		>>> from tests.examples import SomeClass, SomeExtendedClass
113		>>> c = Container()
114		>>> c[SomeClass] = SomeExtendedClass
115		>>> c[SomeClass]
116		<tests.examples.SomeExtendedClass object at ...>
117		"""
118
119	1	_registered_types: Dict[Type, SpecialDepDefinition]
120	1	_parent_definitions: DefinitionsSource
121	1	_reflector: CachingReflector
122	1	_undefined_logger: logging.Logger
123
124	1	def __init__(
125		self,
126		container: Optional["Container"] = None,
127		log_undefined_deps: Union[bool, logging.Logger] = False,
128		):
129		"""
130		:param container: Optional container if provided the existing definitions will be copied
131		:param log_undefined_deps indicates if a log message should be emmited when an undefined dep is loaded
132		"""
133
134		# ContainerDebugInfo is always registered
135		# This means consumers can consume an overview of the container
136		# without hacking anything custom together.
137	1	self._registered_types = {
138		ContainerDebugInfo: ConstructionWithoutContainer(lambda: self)
139		}
140
141	1	if container:
142	1	self._parent_definitions = container
143	1	self._reflector = container._reflector
144		else:
145	1	self._parent_definitions = EmptyDefinitionSet()
146	1	self._reflector = CachingReflector()
147
148	1	if not log_undefined_deps:
149	1	self._undefined_logger = NullLogger()
150	1	elif log_undefined_deps is True:
151	1	self._undefined_logger = logging.getLogger(__name__)
152		else:
153	1	self._undefined_logger = cast(logging.Logger, log_undefined_deps)
154
155	1	def define(self, dep: Type[X], resolver: TypeResolver[X]) -> SpecialDepDefinition:
156		"""Register how to construct an object of type X
157
158		>>> from tests.examples import SomeClass
159		>>> c = Container()
160		>>> c.define(SomeClass, lambda: SomeClass())
161		<lagom.definitions.ConstructionWithoutContainer ...>
162
163		:param dep: The type to be constructed
164		:param resolver: A definition of how to construct it
165		:return:
166		"""
167	1	if dep in UNRESOLVABLE_TYPES:
168		raise InvalidDependencyDefinition()
169	1	if dep in self._registered_types:
170	1	raise DuplicateDefinition()
171	1	if dep is resolver:
172		# This is a special case for things like container[Foo] = Foo
173	1	return self.define(dep, Alias(dep, skip_definitions=True))
174	1	definition = normalise(resolver)
175	1	self._registered_types[dep] = definition
176	1	self._registered_types[Optional[dep]] = definition # type: ignore
177
178		# For awaitables we add a convenience exception to be thrown if code hints on the type
179		# without the awaitable.
180	1	awaitable_type = remove_awaitable_type(dep)
181	1	if awaitable_type:
182		# Unless there's already a sync version defined.
183	1	if awaitable_type not in self.defined_types:
184	1	self._registered_types[awaitable_type] = UnresolvableTypeDefinition(
185		TypeOnlyAvailableAsAwaitable(awaitable_type)
186		)
187	1	return definition
188
189	1	@property
190	1	def defined_types(self) -> Set[Type]:
191		"""The types the container has explicit build instructions for
192
193		:return:
194		"""
195	1	return self._parent_definitions.defined_types.union(
196		self._registered_types.keys()
197		)
198
199	1	@property
200	1	def reflection_cache_overview(self) -> Dict[str, str]:
201	1	return self._reflector.overview_of_cache
202
203	1	def temporary_singletons(
204		self, singletons: Optional[List[Type]] = None
205		) -> "TemporaryInjectionContext":
206		"""
207		Returns a context that loads a new container with singletons that only exist
208		for the context.
209
210		>>> from tests.examples import SomeClass
211		>>> base_container = Container()
212		>>> def my_func():
213		... with base_container.temporary_singletons([SomeClass]) as c:
214		... assert c[SomeClass] is c[SomeClass]
215		>>> my_func()
216
217		:param singletons: items which should be considered singletons within the context
218		:return:
219		"""
220	1	updater = (
221		functools.partial(update_container_singletons, singletons=singletons)
222		if singletons
223		else None
224		)
225	1	return TemporaryInjectionContext(self, updater)
226
227	1	def resolve(
228		self, dep_type: Type[X], suppress_error=False, skip_definitions=False
229		) -> X:
230		"""Constructs an object of type X
231
232		If the object can't be constructed an exception will be raised unless
233		supress errors is true
234
235		>>> from tests.examples import SomeClass
236		>>> c = Container()
237		>>> c.resolve(SomeClass)
238		<tests.examples.SomeClass object at ...>
239
240		>>> from tests.examples import SomeClass
241		>>> c = Container()
242		>>> c.resolve(int)
243		Traceback (most recent call last):
244		...
245		lagom.exceptions.UnresolvableType: ...
246
247		Optional wrappers are stripped out to be what is being asked for
248		>>> from tests.examples import SomeClass
249		>>> c = Container()
250		>>> c.resolve(Optional[SomeClass])
251		<tests.examples.SomeClass object at ...>
252
253		:param dep_type: The type of object to construct
254		:param suppress_error: if true returns None on failure
255		:param skip_definitions:
256		:return:
257		"""
258	1	if not skip_definitions:
259	1	definition = self.get_definition(dep_type)
260	1	if definition:
261	1	return definition.get_instance(self)
262
263	1	optional_dep_type = remove_optional_type(dep_type)
264	1	if optional_dep_type:
265	1	return self.resolve(optional_dep_type, suppress_error=True)
266
267	1	return self._reflection_build_with_err_handling(dep_type, suppress_error)
268
269	1	def partial(
270		self,
271		func: Callable[..., X],
272		shared: Optional[List[Type]] = None,
273		container_updater: Optional[CallTimeContainerUpdate] = None,
274		) -> ContainerBoundFunction[X]:
275		"""Takes a callable and returns a callable bound to the container
276		When invoking the new callable if any arguments have a default set
277		to the special marker object "injectable" then they will be constructed by
278		the container. For automatic injection without the marker use "magic_partial"
279		>>> from tests.examples import SomeClass
280		>>> c = Container()
281		>>> def my_func(something: SomeClass = injectable):
282		... return f"Successfully called with {something}"
283		>>> bound_func = c.magic_partial(my_func)
284		>>> bound_func()
285		'Successfully called with <tests.examples.SomeClass object at ...>'
286
287		:param func: the function to bind to the container
288		:param shared: items which should be considered singletons on a per call level
289		:param container_updater: An optional callable to update the container before resolution
290		:return:
291		"""
292	1	spec = self._get_spec_without_self(func)
293	1	keys_to_bind = (
294		key for (key, arg) in spec.defaults.items() if arg is injectable
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-08-31 11:35 UTC by Report Bug Copy Issue Report The variable `key` does not seem to be defined. Loading history...
295		)
296	1	keys_and_types = [(key, spec.annotations[key]) for key in keys_to_bind]
297
298	1	base_injection_context = self.temporary_singletons(shared)
299	1	update_container = container_updater if container_updater else _update_nothing
300
301	1	def _update_args(_injection_context, supplied_args, supplied_kwargs):
302	1	keys_to_skip = set(supplied_kwargs.keys())
303	1	keys_to_skip.update(spec.args[0 : len(supplied_args)])
304	1	with _injection_context as invocation_container:
305	1	update_container(invocation_container, supplied_args, supplied_kwargs)
306	1	kwargs = {
307		key: invocation_container.resolve(dep_type)
308		for (key, dep_type) in keys_and_types
309		if key not in keys_to_skip
310		}
311	1	kwargs.update(supplied_kwargs)
312	1	return supplied_args, kwargs
313
314	1	return apply_argument_updater(func, base_injection_context, _update_args, spec)
315
316	1	def magic_partial(
317		self,
318		func: Callable[..., X],
319		shared: Optional[List[Type]] = None,
320		keys_to_skip: Optional[List[str]] = None,
321		skip_pos_up_to: int = 0,
322		container_updater: Optional[CallTimeContainerUpdate] = None,
323		) -> ContainerBoundFunction[X]:
324		"""Takes a callable and returns a callable bound to the container
325		When invoking the new callable if any arguments can be constructed by the container
326		then they can be ommited.
327		>>> from tests.examples import SomeClass
328		>>> c = Container()
329		>>> def my_func(something: SomeClass):
330		... return f"Successfully called with {something}"
331		>>> bound_func = c.magic_partial(my_func)
332		>>> bound_func()
333		'Successfully called with <tests.examples.SomeClass object at ...>'
334
335		:param func: the function to bind to the container
336		:param shared: items which should be considered singletons on a per call level
337		:param keys_to_skip: named arguments which the container shouldnt build
338		:param skip_pos_up_to: positional arguments which the container shouldnt build
339		:param container_updater: An optional callable to update the container before resolution
340		:return:
341		"""
342	1	spec = self._get_spec_without_self(func)
343
344	1	update_container = container_updater if container_updater else _update_nothing
345	1	base_injection_context = self.temporary_singletons(shared)
346
347	1	def _update_args(_injection_context, supplied_args, supplied_kwargs):
348	1	final_keys_to_skip = (keys_to_skip or []) + list(supplied_kwargs.keys())
349	1	final_skip_pos_up_to = max(skip_pos_up_to, len(supplied_args))
350	1	with _injection_context as invocation_container:
351	1	update_container(invocation_container, supplied_args, supplied_kwargs)
352	1	kwargs = invocation_container._infer_dependencies(
353		spec,
354		suppress_error=True,
355		keys_to_skip=final_keys_to_skip,
356		skip_pos_up_to=final_skip_pos_up_to,
357		)
358	1	kwargs.update(supplied_kwargs)
359	1	return supplied_args, kwargs
360
361	1	return apply_argument_updater(
362		func, base_injection_context, _update_args, spec, catch_errors=True
363		)
364
365	1	def clone(self) -> "Container":
366		"""returns a copy of the container
367		:return:
368		"""
369	1	return Container(self, log_undefined_deps=self._undefined_logger)
370
371	1	def get_definition(self, dep_type: Type[X]) -> Optional[SpecialDepDefinition[X]]:
372		"""
373		Will return the definition in this container. If none has been defined any
374		definition in the parent container will be used.
375
376		:param dep_type:
377		:return:
378		"""
379	1	definition = self._registered_types.get(dep_type, Unset)
380	1	if definition is Unset:
381	1	return self._parent_definitions.get_definition(dep_type)
382	1	return definition
383
384	1	def __getitem__(self, dep: Type[X]) -> X:
385	1	return self.resolve(dep)
386
387	1	def __setitem__(self, dep: Type[X], resolver: TypeResolver[X]):
388	1	self.define(dep, resolver)
389
390	1	def _reflection_build_with_err_handling(
391		self, dep_type: Type[X], suppress_error: bool
392		) -> X:
393	1	try:
394	1	if dep_type in UNRESOLVABLE_TYPES:
395	1	raise UnresolvableType(dep_type)
396	1	return self._reflection_build(dep_type)
397	1	except UnresolvableType as inner_error:
398	1	if not suppress_error:
399	1	raise UnresolvableType(dep_type) from inner_error
400	1	return None # type: ignore
401	1	except RecursionError as recursion_error:
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-08-31 13:57 UTC by Report Bug Copy Issue Report The variable `RecursionError` does not seem to be defined. Loading history...
402		raise RecursiveDefinitionError(dep_type) from recursion_error
403
404	1	def _reflection_build(self, dep_type: Type[X]) -> X:
405	1	self._undefined_logger.warning(
406		f"Undefined dependency. Using reflection for {dep_type}",
407		extra={"undefined_dependency": dep_type},
408		)
409	1	spec = self._reflector.get_function_spec(dep_type.__init__)
410	1	sub_deps = self._infer_dependencies(spec, types_to_skip={dep_type})
411	1	try:
412	1	return dep_type(**sub_deps) # type: ignore
413	1	except TypeError as type_error:
414	1	raise UnresolvableType(dep_type) from type_error
415
416	1	def _infer_dependencies(
417		self,
418		spec: FunctionSpec,
419		suppress_error=False,
420		keys_to_skip: Optional[List[str]] = None,
421		skip_pos_up_to=0,
422		types_to_skip: Optional[Set[Type]] = None,
423		):
424	1	dep_keys_to_skip: List[str] = []
425	1	dep_keys_to_skip.extend(spec.args[0:skip_pos_up_to])
426	1	dep_keys_to_skip.extend(keys_to_skip or [])
427	1	types_to_skip = types_to_skip or set()
428	1	sub_deps = {
429		key: self.resolve(sub_dep_type, suppress_error=suppress_error)
430		for (key, sub_dep_type) in spec.annotations.items()
431		if sub_dep_type != Any
432		and (key not in dep_keys_to_skip)
433		and (sub_dep_type not in types_to_skip)
434		}
435	1	return {key: dep for (key, dep) in sub_deps.items() if dep is not None}
436
437	1	def _get_spec_without_self(self, func: Callable[..., X]) -> FunctionSpec:
438	1	if isinstance(func, (FunctionType, MethodType)):
439	1	return self._reflector.get_function_spec(func)
440	1	t = cast(Type[X], func)
441	1	return self._reflector.get_function_spec(t.__init__).without_argument("self")
442
443
444	1	@mypyc_attr(allow_interpreted_subclasses=True)
445	1	class ExplicitContainer(Container):
446	1	def resolve(
447		self, dep_type: Type[X], suppress_error=False, skip_definitions=False
448		) -> X:
449	1	definition = self.get_definition(dep_type)
450	1	if not definition:
451	1	if suppress_error:
452	1	return None # type: ignore
453	1	raise DependencyNotDefined(dep_type)
454	1	return definition.get_instance(self)
455
456	1	def define(self, dep, resolver):
457	1	definition = super().define(dep, resolver)
458	1	if isinstance(definition, Alias):
459	1	raise InvalidDependencyDefinition(
460		"Aliases are not valid in an explicit container"
461		)
462	1	if isinstance(definition, Singleton) and isinstance(
463		definition.singleton_type, Alias
464		):
465	1	raise InvalidDependencyDefinition(
466		"Aliases are not valid inside singletons in an explicit container"
467		)
468	1	return definition
469
470	1	def clone(self):
471		"""returns a copy of the container
472		:return:
473		"""
474	1	return ExplicitContainer(self, log_undefined_deps=self._undefined_logger)
475
476
477	1	class EmptyDefinitionSet(DefinitionsSource):
478		"""
479		Represents the starting state for a collection of dependency definitions
480		i.e. None and everything has to be built with reflection
481		"""
482
483	1	def get_definition(self, dep_type: Type[X]) -> Optional[SpecialDepDefinition[X]]:
484		"""
485		No types are defined in the empty set
486		:param dep_type:
487		:return:
488		"""
489	1	return None
490
491	1	@property
492	1	def defined_types(self) -> Set[Type]:
493	1	return set()
494
495
496	1	def _update_nothing(_c: WriteableContainer, _a: typing.Collection, _k: Dict):
497		return None
498

meadsteve / lagom

Pull Request — master (#226)

EmptyDefinitionSet.get_definition() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like