lagom.experimental.context_based.AsyncContextContainer.__init__() - 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 12:32 UTC

AsyncContextContainer.init() A

↳ Parent: lagom.experimental.context_based

Complexity

Conditions

Size

Total Lines	10
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	5
dl	0
loc	10
ccs	4
cts	4
cp	1
crap	1
rs	9.95
c	0
b	0
f	0

import inspect
import logging
from asyncio import Lock
from contextlib import AsyncExitStack
from copy import copy
from typing import (
    Optional,
    Type,
    TypeVar,
    Awaitable,
    Generic,
    Collection,
    Union,
    ContextManager,
    AsyncContextManager,
    Iterator,
    Generator,
    AsyncGenerator,
    Callable,
    List,
    Dict,
)

from lagom.container import Container
from lagom.functools import wraps
from lagom.definitions import Alias, ConstructionWithContainer, SingletonWrapper
from lagom.exceptions import InvalidDependencyDefinition, MissingFeature
from lagom.experimental.definitions import AsyncConstructionWithContainer
from lagom.interfaces import (
    ReadableContainer,
    SpecialDepDefinition,
    CallTimeContainerUpdate,
    ContainerBoundFunction,
)

T = TypeVar("T")
X = TypeVar("X")


class AwaitableSingleton(Generic[T]):
    instance: Optional[T]
    constructor: ConstructionWithContainer[Awaitable[T]]
    container: Container
    _lock: Lock

    def __init__(self, constructor: ConstructionWithContainer, container: Container):
        self.instance = None
        self.constructor = constructor  # type: ignore
        self.container = container
        self._lock = Lock()

    async def get(self) -> T:
        if not self.instance:
            async with self._lock:
                if not self.instance:
                    self.instance = await self.constructor.get_instance(self.container)
        return self.instance


class _AsyncContextBoundFunction(ContainerBoundFunction[X]):

    """
    Represents an instance of a function bound to an async context container
    """

    async_context_container: "AsyncContextContainer"
    partially_bound_function: ContainerBoundFunction
    __dict__: Dict

    __slots__ = ("async_context_container", "partially_bound_function")

    def __init__(
        self,
        async_context_container: "AsyncContextContainer",
        partially_bound_function: ContainerBoundFunction,
    ):
        self.async_context_container = async_context_container
        self.partially_bound_function = partially_bound_function

    def __call__(self, *args, **kwargs) -> X:
        return self.__async_call__(*args, **kwargs)

    async def __async_call__(self, *args, **kwargs):
        async with self.async_context_container as c:
            return await self.partially_bound_function.rebind(c)(*args, **kwargs)

    def rebind(self, container: ReadableContainer) -> "ContainerBoundFunction[X]":
        return wraps(self.partially_bound_function)(
            _AsyncContextBoundFunction(
                self.async_context_container,
                self.partially_bound_function.rebind(container),
            )
        )

    def __getattribute__(self, item):
        if item == "__dict__":
            return {}
        return super().__getattribute__(item)


class AsyncContextContainer(Container):
    async_exit_stack: Optional[AsyncExitStack] = None
    _context_types: Collection[Type]
    _context_singletons: Collection[Type]
    _root_context: bool = True

    def __init__(
        self,
        container: Container,
        context_types: Collection[Type],
        context_singletons: Collection[Type] = tuple(),
        log_undefined_deps: Union[bool, logging.Logger] = False,
    ):
        super().__init__(container, log_undefined_deps)
        self._context_types = set(context_types)
        self._context_singletons = set(context_singletons)

    def clone(self) -> "AsyncContextContainer":
        """returns a copy of the container
        :return:
        """
        return AsyncContextContainer(
            self,
            context_types=self._context_types,
            context_singletons=self._context_singletons,
            log_undefined_deps=self._undefined_logger,
        )

    async def __aenter__(self):
        if not self.async_exit_stack and self._root_context:
            self.async_exit_stack = AsyncExitStack()

        if self.async_exit_stack and self._root_context:
            # All actual context definitions happen on a clone so that there's isolation between invocations
            in_context = self.clone()
            in_context.async_exit_stack = AsyncExitStack()
            in_context._root_context = False

            for dep_type in self._context_types:
                managed_dep = self._context_type_def(dep_type)
                key = Awaitable[dep_type] if isinstance(managed_dep, AsyncConstructionWithContainer) else dep_type  # type: ignore
                in_context[key] = managed_dep  # type: ignore
            for dep_type in self._context_singletons:
                managed_singleton = self._singleton_type_def(dep_type)
                key = AwaitableSingleton[dep_type] if isinstance(managed_singleton, AwaitableSingleton) else dep_type  # type: ignore
                in_context[key] = managed_singleton  # type: ignore

            # The parent context manager keeps track of the inner clone
            await self.async_exit_stack.enter_async_context(in_context)
            return in_context
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.async_exit_stack:
            await self.async_exit_stack.aclose()
            self.async_exit_stack = None

    def partial(
        self,
        func: Callable[..., X],
        shared: Optional[List[Type]] = None,
        container_updater: Optional[CallTimeContainerUpdate] = None,
    ) -> ContainerBoundFunction[X]:
        if not inspect.iscoroutinefunction(func):
            raise MissingFeature(
                "AsyncContextManager currently can only deal with async functions"
            )
        base_partial = super(AsyncContextContainer, self).partial(
            func, shared, container_updater
        )

        return wraps(base_partial)(_AsyncContextBoundFunction(self, base_partial))

    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]:
        if not inspect.iscoroutinefunction(func):
            raise MissingFeature(
                "AsyncContextManager currently can only deal with async functions"
            )
        base_partial = super(AsyncContextContainer, self).magic_partial(
            func, shared, keys_to_skip, skip_pos_up_to, container_updater
        )

        return wraps(base_partial)(_AsyncContextBoundFunction(self, base_partial))

    def _context_type_def(self, dep_type: Type):
        type_def = self.get_definition(ContextManager[dep_type]) or self.get_definition(Iterator[dep_type]) or self.get_definition(Generator[dep_type, None, None]) or self.get_definition(AsyncGenerator[dep_type, None]) or self.get_definition(AsyncContextManager[dep_type])  # type: ignore
        if type_def is None:
            raise InvalidDependencyDefinition(
                f"A ContextManager[{dep_type}] should be defined. "
                f"This could be an Iterator[{dep_type}] or Generator[{dep_type}, None, None] "
                f"with the @contextmanager decorator"
            )
        if isinstance(type_def, Alias):
            # Without this we create a definition that points to
            # itself.
            type_def = copy(type_def)
            type_def.skip_definitions = True
        if self.get_definition(AsyncGenerator[dep_type, None]) or self.get_definition(AsyncContextManager[dep_type]):  # type: ignore
            return AsyncConstructionWithContainer(lambda c: self._async_context_resolver(c, type_def))  # type: ignore
        return ConstructionWithContainer(lambda c: self._context_resolver(c, type_def))  # type: ignore

    def _context_resolver(self, c: ReadableContainer, type_def: SpecialDepDefinition):
        """
        Takes an existing definition which must be a context manager. Returns
        the value of the context manager from __enter__ and then places the
        __exit__ in this container's exit stack
        """
        assert self.async_exit_stack, "Types can only be resolved within an async with"
        context_manager = type_def.get_instance(c)
        return self.async_exit_stack.enter_context(context_manager)

    def _async_context_resolver(
        self, c: ReadableContainer, type_def: SpecialDepDefinition
    ):
        """
        Takes an existing definition which must be a context manager. Returns
        the value of the context manager from __aenter__ and then places the
        __aexit__ in this container's exit stack
        """
        assert self.async_exit_stack, "Types can only be resolved within an async with"
        context_manager = type_def.get_instance(c)
        return self.async_exit_stack.enter_async_context(context_manager)

    def _singleton_type_def(self, dep_type: Type):
        """
        The same as context_type_def but acts as a singleton within this container
        """
        type_def = self._context_type_def(dep_type)
        if isinstance(type_def, AsyncConstructionWithContainer):
            return AwaitableSingleton(type_def, self)
        return SingletonWrapper(type_def)


1	1		import inspect
2	1		import logging
3	1		from asyncio import Lock
4	1		from contextlib import AsyncExitStack
5	1		from copy import copy
6	1		from typing import (
7			Optional,
8			Type,
9			TypeVar,
10			Awaitable,
11			Generic,
12			Collection,
13			Union,
14			ContextManager,
15			AsyncContextManager,
16			Iterator,
17			Generator,
18			AsyncGenerator,
19			Callable,
20			List,
21			Dict,
22			)
23
24	1		from lagom.container import Container
25	1		from lagom.functools import wraps
26	1		from lagom.definitions import Alias, ConstructionWithContainer, SingletonWrapper
27	1		from lagom.exceptions import InvalidDependencyDefinition, MissingFeature
28	1		from lagom.experimental.definitions import AsyncConstructionWithContainer
29	1		from lagom.interfaces import (
30			ReadableContainer,
31			SpecialDepDefinition,
32			CallTimeContainerUpdate,
33			ContainerBoundFunction,
34			)
35
36	1		T = TypeVar("T")
37	1		X = TypeVar("X")
38
39
40	1		class AwaitableSingleton(Generic[T]):
41	1		instance: Optional[T]
42	1		constructor: ConstructionWithContainer[Awaitable[T]]
43	1		container: Container
44	1		_lock: Lock
45
46	1		def __init__(self, constructor: ConstructionWithContainer, container: Container):
47	1		self.instance = None
48	1		self.constructor = constructor # type: ignore
49	1		self.container = container
50	1		self._lock = Lock()
51
52	1		async def get(self) -> T:
53	1		if not self.instance:
54	1		async with self._lock:
55	1		if not self.instance:
56	1		self.instance = await self.constructor.get_instance(self.container)
57	1		return self.instance
58
59
60	1	View Code Duplication	class _AsyncContextBoundFunction(ContainerBoundFunction[X]):
			0 ignored issues – show Duplication introduced 2023-03-06 15:27 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
61			"""
62			Represents an instance of a function bound to an async context container
63			"""
64
65	1		async_context_container: "AsyncContextContainer"
66	1		partially_bound_function: ContainerBoundFunction
67	1		__dict__: Dict
68
69	1		__slots__ = ("async_context_container", "partially_bound_function")
70
71	1		def __init__(
72			self,
73			async_context_container: "AsyncContextContainer",
74			partially_bound_function: ContainerBoundFunction,
75			):
76	1		self.async_context_container = async_context_container
77	1		self.partially_bound_function = partially_bound_function
78
79	1		def __call__(self, args, *kwargs) -> X:
80	1		return self.__async_call__(args, *kwargs)
81
82	1		async def __async_call__(self, args, *kwargs):
83	1		async with self.async_context_container as c:
84	1		return await self.partially_bound_function.rebind(c)(args, *kwargs)
85
86	1		def rebind(self, container: ReadableContainer) -> "ContainerBoundFunction[X]":
87			return wraps(self.partially_bound_function)(
88			_AsyncContextBoundFunction(
89			self.async_context_container,
90			self.partially_bound_function.rebind(container),
91			)
92			)
93
94	1		def __getattribute__(self, item):
95	1		if item == "__dict__":
96			return {}
97	1		return super().__getattribute__(item)
98
99
100	1		class AsyncContextContainer(Container):
101	1		async_exit_stack: Optional[AsyncExitStack] = None
102	1		_context_types: Collection[Type]
103	1		_context_singletons: Collection[Type]
104	1		_root_context: bool = True
105
106	1		def __init__(
107			self,
108			container: Container,
109			context_types: Collection[Type],
110			context_singletons: Collection[Type] = tuple(),
111			log_undefined_deps: Union[bool, logging.Logger] = False,
112			):
113	1		super().__init__(container, log_undefined_deps)
114	1		self._context_types = set(context_types)
115	1		self._context_singletons = set(context_singletons)
116
117	1		def clone(self) -> "AsyncContextContainer":
118			"""returns a copy of the container
119			:return:
120			"""
121	1		return AsyncContextContainer(
122			self,
123			context_types=self._context_types,
124			context_singletons=self._context_singletons,
125			log_undefined_deps=self._undefined_logger,
126			)
127
128	1		async def __aenter__(self):
129	1		if not self.async_exit_stack and self._root_context:
130	1		self.async_exit_stack = AsyncExitStack()
131
132	1		if self.async_exit_stack and self._root_context:
133			# All actual context definitions happen on a clone so that there's isolation between invocations
134	1		in_context = self.clone()
135	1		in_context.async_exit_stack = AsyncExitStack()
136	1		in_context._root_context = False
137
138	1		for dep_type in self._context_types:
139	1		managed_dep = self._context_type_def(dep_type)
140	1		key = Awaitable[dep_type] if isinstance(managed_dep, AsyncConstructionWithContainer) else dep_type # type: ignore
141	1		in_context[key] = managed_dep # type: ignore
142	1		for dep_type in self._context_singletons:
143	1		managed_singleton = self._singleton_type_def(dep_type)
144	1		key = AwaitableSingleton[dep_type] if isinstance(managed_singleton, AwaitableSingleton) else dep_type # type: ignore
145	1		in_context[key] = managed_singleton # type: ignore
146
147			# The parent context manager keeps track of the inner clone
148	1		await self.async_exit_stack.enter_async_context(in_context)
149	1		return in_context
150	1		return self
151
152	1		async def __aexit__(self, exc_type, exc_val, exc_tb):
153	1		if self.async_exit_stack:
154	1		await self.async_exit_stack.aclose()
155	1		self.async_exit_stack = None
156
157	1		def partial(
158			self,
159			func: Callable[..., X],
160			shared: Optional[List[Type]] = None,
161			container_updater: Optional[CallTimeContainerUpdate] = None,
162			) -> ContainerBoundFunction[X]:
163	1		if not inspect.iscoroutinefunction(func):
164			raise MissingFeature(
165			"AsyncContextManager currently can only deal with async functions"
166			)
167	1		base_partial = super(AsyncContextContainer, self).partial(
168			func, shared, container_updater
169			)
170
171	1		return wraps(base_partial)(_AsyncContextBoundFunction(self, base_partial))
172
173	1		def magic_partial(
174			self,
175			func: Callable[..., X],
176			shared: Optional[List[Type]] = None,
177			keys_to_skip: Optional[List[str]] = None,
178			skip_pos_up_to: int = 0,
179			container_updater: Optional[CallTimeContainerUpdate] = None,
180			) -> ContainerBoundFunction[X]:
181	1		if not inspect.iscoroutinefunction(func):
182			raise MissingFeature(
183			"AsyncContextManager currently can only deal with async functions"
184			)
185	1		base_partial = super(AsyncContextContainer, self).magic_partial(
186			func, shared, keys_to_skip, skip_pos_up_to, container_updater
187			)
188
189	1		return wraps(base_partial)(_AsyncContextBoundFunction(self, base_partial))
190
191	1		def _context_type_def(self, dep_type: Type):
192	1		type_def = self.get_definition(ContextManager[dep_type]) or self.get_definition(Iterator[dep_type]) or self.get_definition(Generator[dep_type, None, None]) or self.get_definition(AsyncGenerator[dep_type, None]) or self.get_definition(AsyncContextManager[dep_type]) # type: ignore
193	1		if type_def is None:
194	1		raise InvalidDependencyDefinition(
195			f"A ContextManager[{dep_type}] should be defined. "
196			f"This could be an Iterator[{dep_type}] or Generator[{dep_type}, None, None] "
197			f"with the @contextmanager decorator"
198			)
199	1		if isinstance(type_def, Alias):
200			# Without this we create a definition that points to
201			# itself.
202	1		type_def = copy(type_def)
203	1		type_def.skip_definitions = True
204	1		if self.get_definition(AsyncGenerator[dep_type, None]) or self.get_definition(AsyncContextManager[dep_type]): # type: ignore
205	1		return AsyncConstructionWithContainer(lambda c: self._async_context_resolver(c, type_def)) # type: ignore
206	1		return ConstructionWithContainer(lambda c: self._context_resolver(c, type_def)) # type: ignore
207
208	1		def _context_resolver(self, c: ReadableContainer, type_def: SpecialDepDefinition):
209			"""
210			Takes an existing definition which must be a context manager. Returns
211			the value of the context manager from __enter__ and then places the
212			__exit__ in this container's exit stack
213			"""
214	1		assert self.async_exit_stack, "Types can only be resolved within an async with"
215	1		context_manager = type_def.get_instance(c)
216	1		return self.async_exit_stack.enter_context(context_manager)
217
218	1		def _async_context_resolver(
219			self, c: ReadableContainer, type_def: SpecialDepDefinition
220			):
221			"""
222			Takes an existing definition which must be a context manager. Returns
223			the value of the context manager from __aenter__ and then places the
224			__aexit__ in this container's exit stack
225			"""
226	1		assert self.async_exit_stack, "Types can only be resolved within an async with"
227	1		context_manager = type_def.get_instance(c)
228	1		return self.async_exit_stack.enter_async_context(context_manager)
229
230	1		def _singleton_type_def(self, dep_type: Type):
231			"""
232			The same as context_type_def but acts as a singleton within this container
233			"""
234	1		type_def = self._context_type_def(dep_type)
235	1		if isinstance(type_def, AsyncConstructionWithContainer):
236	1		return AwaitableSingleton(type_def, self)
237			return SingletonWrapper(type_def)
238

meadsteve / lagom

Pull Request — master (#226)

AsyncContextContainer.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

AsyncContextContainer.init() A