-
Notifications
You must be signed in to change notification settings - Fork 2
/
middlewares.py
56 lines (44 loc) · 1.72 KB
/
middlewares.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import typing
from . import exceptions, types
class CircularDependencyCircuitBreaker(types.Middleware):
def __init__(self):
self._requested_types: typing.Dict[types.DependencyType, int] = {}
self._stack: typing.List[types.DependencyType] = []
def execute(
self,
dependency_type: types.DependencyType,
next_middleware: types.NextMiddleware,
):
self._raise_if_circular_dependency(dependency_type)
self._track(dependency_type)
try:
return next_middleware()
finally:
self._untrack(dependency_type)
def _raise_if_circular_dependency(self, dependency_type: types.DependencyType):
if dependency_type in self._requested_types:
cycle = self._stack[self._requested_types[dependency_type] :] + [
dependency_type
]
raise exceptions.CircularDependency(cycle=cycle)
def _track(self, dependency_type: types.DependencyType):
self._requested_types[dependency_type] = len(self._stack)
self._stack.append(dependency_type)
def _untrack(self, dependency_type):
del self._requested_types[dependency_type]
self._stack.pop()
class Singleton(types.Middleware):
def __init__(self):
self._dependency_values: typing.Dict[types.DependencyType, typing.Any] = {}
def execute(
self,
dependency_type: types.DependencyType,
next_middleware: types.NextMiddleware,
):
try:
return self._dependency_values[dependency_type]
except KeyError:
pass
dependency_value = next_middleware()
self._dependency_values[dependency_type] = dependency_value
return dependency_value