lagom.container.Container.magic_partial() - Code Metrics - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

lagom.container.Container.magic_partial() A
last analyzed 2024-11-22 07:23 UTC

↳ Parent: lagom.container

Complexity

Conditions

Size

Total Lines	46
Code Lines	23

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	3

Importance

Changes

Metric	Value
cc	3
eloc	23
nop	6
dl	0
loc	46
ccs	13
cts	13
cp	1
crap	3
rs	9.328
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 .interfaces import (
    SpecialDepDefinition,
    WriteableContainer,
    TypeResolver,
    DefinitionsSource,
    ExtendableContainer,
    ContainerDebugInfo,
    CallTimeContainerUpdate,
)
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), 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,
    ) -> Callable[..., 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]

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

        def _update_args(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, _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,
    ) -> Callable[..., 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
        _injection_context = self.temporary_singletons(shared)

        def _update_args(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, _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()


class _TemporaryInjectionContext:
    _base_container: Container

    def __init__(
        self,
        container: Container,
        update_function: Optional[Callable[[Container], Container]] = None,
    ):
        self._base_container = container
        if update_function:
            self._build_temporary_container = lambda: update_function(
                self._base_container
            )
        else:
            self._build_temporary_container = lambda: self._base_container.clone()

    def __enter__(self) -> Container:
        return self._build_temporary_container()

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


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

meadsteve / lagom

lagom.container.Container.magic_partial() A last analyzed 2024-11-22 07:23 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

lagom.container.Container.magic_partial() A
last analyzed 2024-11-22 07:23 UTC