lagom.definitions.ConstructionWithoutContainer.__init__() - Code Metrics - Inspection of "Fix bug where context singletons existed after cle..." - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — async-context-container (98d6fa)

by Steve

created 2022-11-07 18:49 UTC

ConstructionWithoutContainer.init() A

↳ 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 inspect
from threading import Lock
from typing import Union, Type, Optional, Callable, TypeVar, NoReturn, Iterator

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

X = TypeVar("X")


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 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()

    def reset(self):
        try:
            self._thread_lock.acquire()
            self._instance = None
        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
    """

    _msg_or_exception: Union[str, Exception]

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

meadsteve / lagom

Branch — async-context-container (98d6fa)

ConstructionWithoutContainer.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

ConstructionWithoutContainer.init() A