Total Complexity | 40 |
Total Lines | 238 |
Duplicated Lines | 15.97 % |
Coverage | 96.52% |
Changes | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like lagom.experimental.context_based often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
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]): |
|
|||
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 |