lagom.definitions.AsyncConstructionWithContainer.get_instance() - Code Metrics - Inspection of "Add ability to make singletons out of async calls..." - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e89298...4f440b )

by Steve

created 2023-09-18 15:01 UTC

AsyncConstructionWithContainer.get_instance() A

↳ Parent: lagom.definitions

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.2963

Importance

Changes

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

"""
Classes representing specific ways of representing dependencies
"""
import asyncio
import inspect
from threading import Lock
from typing import (
    Union,
    Type,
    Optional,
    Callable,
    TypeVar,
    NoReturn,
    Iterator,
    Awaitable,
)

from .exceptions import (
    InvalidDependencyDefinition,
    TypeResolutionBlocked,
)
from .interfaces import SpecialDepDefinition, ReadableContainer, TypeResolver
from .util.functional import arity

X = TypeVar("X")
AX = Awaitable[X]


class AsyncConstructionWithoutContainer(SpecialDepDefinition[AX]):
    """Wraps an awaitable for constructing a type"""

    def __init__(self, constructor: Callable[[], AX]):
        self.constructor = constructor

    def get_instance(self, container: ReadableContainer) -> AX:
        resolver = self.constructor
        return asyncio.ensure_future(resolver())


class AsyncConstructionWithContainer(SpecialDepDefinition[AX]):
    """Wraps an awaitable for constructing a type"""

    def __init__(self, constructor: Callable[[ReadableContainer], AX]):
        self.constructor = constructor

    def get_instance(self, container: ReadableContainer) -> Awaitable[AX]:
        resolver = self.constructor
        return asyncio.ensure_future(resolver(container))


class ConstructionWithoutContainer(SpecialDepDefinition[X]):
    """Wraps a callable for constructing a type"""

    def __init__(self, constructor: Callable[[], X]):
        self.constructor = constructor

    def get_instance(self, container: ReadableContainer) -> X:
        resolver = self.constructor
        return resolver()


class ConstructionWithContainer(SpecialDepDefinition[X]):
    """Wraps a callable for constructing a type"""

    def __init__(self, constructor: Callable[[ReadableContainer], X]):
        self.constructor = constructor

    def get_instance(self, container: ReadableContainer) -> X:
        resolver = self.constructor
        return resolver(container)


class YieldWithoutContainer(SpecialDepDefinition[X]):
    """Wraps a callable for constructing a type"""

    def __init__(self, constructor: Callable[[], Iterator[X]]):
        self.constructor = constructor

    def get_instance(self, container: ReadableContainer) -> X:
        resolver = self.constructor
        return next(resolver())


class YieldWithContainer(SpecialDepDefinition[X]):
    """Wraps a generator for constructing a type"""

    def __init__(self, constructor: Callable[[ReadableContainer], Iterator[X]]):
        self.constructor = constructor

    def get_instance(self, container: ReadableContainer) -> X:
        resolver = self.constructor
        return next(resolver(container))


def async_construction(
    resolver: Callable,
) -> Union[AsyncConstructionWithContainer, AsyncConstructionWithoutContainer]:
    """
    Takes a generator and returns a type definition
    :param reflector:
    :param resolver:
    :return:
    """
    func_arity = arity(resolver)
    if func_arity == 0:
        return AsyncConstructionWithoutContainer(resolver)
    if func_arity == 1:
        return AsyncConstructionWithContainer(resolver)
    raise InvalidDependencyDefinition(f"Arity {func_arity} functions are not supported")


def construction(
    resolver: Callable,
) -> Union[ConstructionWithContainer, ConstructionWithoutContainer]:
    """
    Takes a generator and returns a type definition
    :param reflector:
    :param resolver:
    :return:
    """
    func_arity = arity(resolver)
    if func_arity == 0:
        return ConstructionWithoutContainer(resolver)
    if func_arity == 1:
        return ConstructionWithContainer(resolver)
    raise InvalidDependencyDefinition(f"Arity {func_arity} functions are not supported")


def yielding_construction(
    resolver: Callable,
) -> Union[YieldWithContainer, YieldWithoutContainer]:
    """
    Takes a generator and returns a type definition
    :param reflector:
    :param resolver:
    :return:
    """
    func_arity = arity(resolver)
    if func_arity == 0:
        return YieldWithoutContainer(resolver)
    if func_arity == 1:
        return YieldWithContainer(resolver)
    raise InvalidDependencyDefinition(
        f"Arity {func_arity} generators are not supported"
    )


class Alias(SpecialDepDefinition[X]):
    """When one class is asked for the other should be returned"""

    alias_type: Type[X]
    skip_definitions: bool

    def __init__(self, alias_type, skip_definitions=False):
        self.alias_type = alias_type
        self.skip_definitions = skip_definitions

    def get_instance(self, container: ReadableContainer) -> X:
        return container.resolve(
            self.alias_type, skip_definitions=self.skip_definitions
        )

    def __copy__(self):
        return Alias(self.alias_type, self.skip_definitions)


class SingletonWrapper(SpecialDepDefinition[X]):
    """Builds only once then saves the built instance"""

    singleton_type: SpecialDepDefinition
    _instance: Optional[X]
    _thread_lock: Lock

    def __init__(self, def_to_wrap: SpecialDepDefinition):
        self.singleton_type = def_to_wrap
        self._instance = None
        self._thread_lock = Lock()

    def get_instance(self, container: ReadableContainer) -> X:
        if self._has_instance:
            return self._instance  # type: ignore
        return self._load_instance(container)

    @property
    def _has_instance(self) -> bool:
        return self._instance is not None

    def _load_instance(self, container):
        try:
            self._thread_lock.acquire()
            if self._has_instance:
                return self._instance
            self._instance = self.singleton_type.get_instance(container)
            return self._instance
        finally:
            self._thread_lock.release()


class Singleton(SingletonWrapper[X]):
    """Builds only once then saves the built instance"""

    def __init__(self, singleton_type: TypeResolver):
        super().__init__(normalise(singleton_type, skip_alias_definitions=True))


class PlainInstance(SpecialDepDefinition[X]):
    """Wraps an actual object that should just be returned"""

    value: X

    def __init__(self, value):
        self.value = value

    def get_instance(self, _) -> X:
        return self.value


class UnresolvableTypeDefinition(SpecialDepDefinition[NoReturn]):
    """
    Used to represent a type that should not be built by the container
    """

    _dep_type: Optional[Type]
    _msg_or_exception: Union[str, Exception]

    def __init__(
        self, msg_or_exception: Union[str, Exception], dep_type: Optional[Type] = None
    ):
        self._dep_type = dep_type
        self._msg_or_exception = msg_or_exception

    def get_instance(self, container: ReadableContainer) -> NoReturn:
        if isinstance(self._msg_or_exception, Exception):
            raise self._msg_or_exception
        else:
            raise TypeResolutionBlocked(self._dep_type, self._msg_or_exception)


def normalise(
    resolver: TypeResolver, skip_alias_definitions=False
) -> SpecialDepDefinition:
    """
    :param resolver:
    :param skip_alias_definitions if an alias is loaded should futher definitions be skipped
    :return:
    """
    if isinstance(resolver, SpecialDepDefinition):
        return resolver
    elif inspect.isfunction(resolver):
        return construction(resolver)  # type: ignore
    elif inspect.iscoroutinefunction(resolver):
        return async_construction(resolver)
    elif inspect.isclass(resolver):
        return Alias(resolver, skip_alias_definitions)
    else:
        return PlainInstance(resolver)


1		"""
2		Classes representing specific ways of representing dependencies
3		"""
4	1	import asyncio
5	1	import inspect
6	1	from threading import Lock
7	1	from typing import (
8		Union,
9		Type,
10		Optional,
11		Callable,
12		TypeVar,
13		NoReturn,
14		Iterator,
15		Awaitable,
16		)
17
18	1	from .exceptions import (
19		InvalidDependencyDefinition,
20		TypeResolutionBlocked,
21		)
22	1	from .interfaces import SpecialDepDefinition, ReadableContainer, TypeResolver
23	1	from .util.functional import arity
24
25	1	X = TypeVar("X")
26	1	AX = Awaitable[X]
27
28
29	1	class AsyncConstructionWithoutContainer(SpecialDepDefinition[AX]):
30		"""Wraps an awaitable for constructing a type"""
31
32	1	def __init__(self, constructor: Callable[[], AX]):
33	1	self.constructor = constructor
34
35	1	def get_instance(self, container: ReadableContainer) -> AX:
36	1	resolver = self.constructor
37	1	return asyncio.ensure_future(resolver())
38
39
40	1	class AsyncConstructionWithContainer(SpecialDepDefinition[AX]):
41		"""Wraps an awaitable for constructing a type"""
42
43	1	def __init__(self, constructor: Callable[[ReadableContainer], AX]):
44		self.constructor = constructor
45
46	1	def get_instance(self, container: ReadableContainer) -> Awaitable[AX]:
47		resolver = self.constructor
48		return asyncio.ensure_future(resolver(container))
49
50
51	1	class ConstructionWithoutContainer(SpecialDepDefinition[X]):
52		"""Wraps a callable for constructing a type"""
53
54	1	def __init__(self, constructor: Callable[[], X]):
55	1	self.constructor = constructor
56
57	1	def get_instance(self, container: ReadableContainer) -> X:
58	1	resolver = self.constructor
59	1	return resolver()
60
61
62	1	class ConstructionWithContainer(SpecialDepDefinition[X]):
63		"""Wraps a callable for constructing a type"""
64
65	1	def __init__(self, constructor: Callable[[ReadableContainer], X]):
66	1	self.constructor = constructor
67
68	1	def get_instance(self, container: ReadableContainer) -> X:
69	1	resolver = self.constructor
70	1	return resolver(container)
71
72
73	1	class YieldWithoutContainer(SpecialDepDefinition[X]):
74		"""Wraps a callable for constructing a type"""
75
76	1	def __init__(self, constructor: Callable[[], Iterator[X]]):
77	1	self.constructor = constructor
78
79	1	def get_instance(self, container: ReadableContainer) -> X:
80	1	resolver = self.constructor
81	1	return next(resolver())
82
83
84	1	class YieldWithContainer(SpecialDepDefinition[X]):
85		"""Wraps a generator for constructing a type"""
86
87	1	def __init__(self, constructor: Callable[[ReadableContainer], Iterator[X]]):
88	1	self.constructor = constructor
89
90	1	def get_instance(self, container: ReadableContainer) -> X:
91	1	resolver = self.constructor
92	1	return next(resolver(container))
93
94
95	1	def async_construction(
96		resolver: Callable,
97		) -> Union[AsyncConstructionWithContainer, AsyncConstructionWithoutContainer]:
98		"""
99		Takes a generator and returns a type definition
100		:param reflector:
101		:param resolver:
102		:return:
103		"""
104	1	func_arity = arity(resolver)
105	1	if func_arity == 0:
106	1	return AsyncConstructionWithoutContainer(resolver)
107		if func_arity == 1:
108		return AsyncConstructionWithContainer(resolver)
109		raise InvalidDependencyDefinition(f"Arity {func_arity} functions are not supported")
110
111
112	1	def construction(
113		resolver: Callable,
114		) -> Union[ConstructionWithContainer, ConstructionWithoutContainer]:
115		"""
116		Takes a generator and returns a type definition
117		:param reflector:
118		:param resolver:
119		:return:
120		"""
121	1	func_arity = arity(resolver)
122	1	if func_arity == 0:
123	1	return ConstructionWithoutContainer(resolver)
124	1	if func_arity == 1:
125	1	return ConstructionWithContainer(resolver)
126	1	raise InvalidDependencyDefinition(f"Arity {func_arity} functions are not supported")
127
128
129	1	def yielding_construction(
130		resolver: Callable,
131		) -> Union[YieldWithContainer, YieldWithoutContainer]:
132		"""
133		Takes a generator and returns a type definition
134		:param reflector:
135		:param resolver:
136		:return:
137		"""
138	1	func_arity = arity(resolver)
139	1	if func_arity == 0:
140	1	return YieldWithoutContainer(resolver)
141	1	if func_arity == 1:
142	1	return YieldWithContainer(resolver)
143		raise InvalidDependencyDefinition(
144		f"Arity {func_arity} generators are not supported"
145		)
146
147
148	1	class Alias(SpecialDepDefinition[X]):
149		"""When one class is asked for the other should be returned"""
150
151	1	alias_type: Type[X]
152	1	skip_definitions: bool
153
154	1	def __init__(self, alias_type, skip_definitions=False):
155	1	self.alias_type = alias_type
156	1	self.skip_definitions = skip_definitions
157
158	1	def get_instance(self, container: ReadableContainer) -> X:
159	1	return container.resolve(
160		self.alias_type, skip_definitions=self.skip_definitions
161		)
162
163	1	def __copy__(self):
164	1	return Alias(self.alias_type, self.skip_definitions)
165
166
167	1	class SingletonWrapper(SpecialDepDefinition[X]):
168		"""Builds only once then saves the built instance"""
169
170	1	singleton_type: SpecialDepDefinition
171	1	_instance: Optional[X]
172	1	_thread_lock: Lock
173
174	1	def __init__(self, def_to_wrap: SpecialDepDefinition):
175	1	self.singleton_type = def_to_wrap
176	1	self._instance = None
177	1	self._thread_lock = Lock()
178
179	1	def get_instance(self, container: ReadableContainer) -> X:
180	1	if self._has_instance:
181	1	return self._instance # type: ignore
182	1	return self._load_instance(container)
183
184	1	@property
185	1	def _has_instance(self) -> bool:
186	1	return self._instance is not None
187
188	1	def _load_instance(self, container):
189	1	try:
190	1	self._thread_lock.acquire()
191	1	if self._has_instance:
192		return self._instance
193	1	self._instance = self.singleton_type.get_instance(container)
194	1	return self._instance
195		finally:
196	1	self._thread_lock.release()
197
198
199	1	class Singleton(SingletonWrapper[X]):
200		"""Builds only once then saves the built instance"""
201
202	1	def __init__(self, singleton_type: TypeResolver):
203	1	super().__init__(normalise(singleton_type, skip_alias_definitions=True))
204
205
206	1	class PlainInstance(SpecialDepDefinition[X]):
207		"""Wraps an actual object that should just be returned"""
208
209	1	value: X
210
211	1	def __init__(self, value):
212	1	self.value = value
213
214	1	def get_instance(self, _) -> X:
215	1	return self.value
216
217
218	1	class UnresolvableTypeDefinition(SpecialDepDefinition[NoReturn]):
219		"""
220		Used to represent a type that should not be built by the container
221		"""
222
223	1	_dep_type: Optional[Type]
224	1	_msg_or_exception: Union[str, Exception]
225
226	1	def __init__(
227		self, msg_or_exception: Union[str, Exception], dep_type: Optional[Type] = None
228		):
229	1	self._dep_type = dep_type
230	1	self._msg_or_exception = msg_or_exception
231
232	1	def get_instance(self, container: ReadableContainer) -> NoReturn:
233	1	if isinstance(self._msg_or_exception, Exception):
234	1	raise self._msg_or_exception
235		else:
236	1	raise TypeResolutionBlocked(self._dep_type, self._msg_or_exception)
237
238
239	1	def normalise(
240		resolver: TypeResolver, skip_alias_definitions=False
241		) -> SpecialDepDefinition:
242		"""
243		:param resolver:
244		:param skip_alias_definitions if an alias is loaded should futher definitions be skipped
245		:return:
246		"""
247	1	if isinstance(resolver, SpecialDepDefinition):
248	1	return resolver
249	1	elif inspect.isfunction(resolver):
250	1	return construction(resolver) # type: ignore
251	1	elif inspect.iscoroutinefunction(resolver):
252	1	return async_construction(resolver)
253	1	elif inspect.isclass(resolver):
254	1	return Alias(resolver, skip_alias_definitions)
255		else:
256		return PlainInstance(resolver)
257

meadsteve / lagom

Push — master ( e89298...4f440b )

AsyncConstructionWithContainer.get_instance() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like