lagom.definitions.PlainInstance.get_instance() - Code Metrics - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

lagom.definitions.PlainInstance.get_instance() A
last analyzed 2024-11-22 07:23 UTC

↳ Parent: lagom.definitions

Complexity

Conditions

Size

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

meadsteve / lagom

lagom.definitions.PlainInstance.get_instance() A last analyzed 2024-11-22 07:23 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

lagom.definitions.PlainInstance.get_instance() A
last analyzed 2024-11-22 07:23 UTC