Passed
Branch async-context-container (98d6fa)
by Steve
03:16
created

lagom.decorators   A

Complexity

Total Complexity 16

Size/Duplication

Total Lines 161
Duplicated Lines 0 %

Test Coverage

Coverage 98.08%

Importance

Changes 0
Metric Value
eloc 72
dl 0
loc 161
rs 10
c 0
b 0
f 0
ccs 51
cts 52
cp 0.9808
wmc 16

6 Functions

Rating   Name   Duplication   Size   Complexity  
A bind_to_container() 0 9 2
A context_dependency_definition() 0 39 5
A _extract_definition_func_and_type() 0 21 4
A dependency_definition() 0 25 2
A magic_bind_to_container() 0 21 2
A _generator_type() 0 2 1
1
"""
2
This module provides decorators for hooking an
3
application into the container.s
4
"""
5 1
import inspect
6 1
from contextlib import contextmanager, asynccontextmanager
7 1
from functools import wraps
8 1
from types import FunctionType
9 1
from typing import (
10
    List,
11
    Type,
12
    Callable,
13
    Tuple,
14
    TypeVar,
15
    ContextManager,
16
    AsyncContextManager,
17
)
18
19 1
from .container import Container
20 1
from .definitions import Singleton, construction, yielding_construction
21 1
from .exceptions import (
22
    MissingReturnType,
23
    ClassesCannotBeDecorated,
24
    InvalidDependencyDefinition,
25
)
26 1
from .interfaces import SpecialDepDefinition
27 1
from .util.reflection import reflect
28
29 1
T = TypeVar("T")
30 1
R = TypeVar("R")
31
32
33 1
def bind_to_container(
34
    container: Container, shared: List[Type] = None
35
) -> Callable[[Callable[..., R]], Callable[..., R]]:
36 1
    def _decorator(func):
37 1
        if not isinstance(func, FunctionType):
38 1
            raise ClassesCannotBeDecorated()
39 1
        return wraps(func)(container.partial(func, shared=shared))
40
41 1
    return _decorator
42
43
44 1
def magic_bind_to_container(
45
    container: Container, shared: List[Type] = None
46
) -> Callable[[Callable[..., R]], Callable[..., R]]:
47
    """Decorates the function so that it's uses the container to construct things
48
49
    >>> from tests.examples import SomeClass
50
    >>> c = Container()
51
    >>> @magic_bind_to_container(c)
52
    ... def say_hello_from(sayer: SomeClass):
53
    ...    return f"hello from {sayer}"
54
    >>> say_hello_from()
55
    'hello from <tests.examples.SomeClass object at ...>'
56
57
    """
58
59 1
    def _decorator(func):
60 1
        if not isinstance(func, FunctionType):
61 1
            raise ClassesCannotBeDecorated()
62 1
        return wraps(func)(container.magic_partial(func, shared=shared))
63
64 1
    return _decorator
65
66
67 1
def dependency_definition(container: Container, singleton: bool = False):
68
    """Registers the provided function with the container
69
    The return type of the decorated function will be reflected and whenever
70
    the container is asked for this type the function will be called
71
72
    >>> from tests.examples import SomeClass, SomeExtendedClass
73
    >>> c = Container()
74
    >>> @dependency_definition(c)
75
    ... def build_some_class_but_extended() -> SomeClass:
76
    ...    return SomeExtendedClass()
77
    >>> c.resolve(SomeClass)
78
    <tests.examples.SomeExtendedClass object at ...>
79
80
    """
81
82 1
    def _decorator(func):
83 1
        definition_func, return_type = _extract_definition_func_and_type(func)  # type: ignore
84
85 1
        if singleton:
86 1
            container.define(return_type, Singleton(definition_func))
87
        else:
88 1
            container.define(return_type, definition_func)
89 1
        return func
90
91 1
    return _decorator
92
93
94 1
def context_dependency_definition(container: Container):
95
    """
96
    Turns the decorated function into a definition for a context manager
97
    in the given container.
98
99
    >>> from tests.examples import SomeClass
100
    >>> from typing import Iterator
101
    >>>
102
    >>> c = Container()
103
    >>>
104
    >>> @context_dependency_definition(c)
105
    ... def my_constructor() -> Iterator[SomeClass]:
106
    ...     try:
107
    ...         yield SomeClass()
108
    ...     finally:
109
    ...         pass # Any tidy up or resource closing could happen here
110
    >>> with c[ContextManager[SomeClass]] as something:
111
    ...     something
112
    <tests.examples.SomeClass ...>
113
114
    with container[ContextManager[MyComplexDep]] as dep:  # type: ignore
115
        assert dep.some_number == 3
116
    """
117
118 1
    def _decorator(func):
119 1
        if not inspect.isgeneratorfunction(func) and not inspect.isasyncgenfunction(
120
            func
121
        ):
122
            raise InvalidDependencyDefinition(
123
                "context_dependency_definition must be given a generator"
124
            )
125 1
        dep_type = _generator_type(reflect(func).return_type)
126 1
        if inspect.isgeneratorfunction(func):
127 1
            container.define(ContextManager[dep_type], contextmanager(func))  # type: ignore
128 1
        if inspect.isasyncgenfunction(func):
129 1
            container.define(AsyncContextManager[dep_type], asynccontextmanager(func))  # type: ignore
130 1
        return func
131
132 1
    return _decorator
133
134
135 1
def _extract_definition_func_and_type(
136
    func,
137
) -> Tuple[SpecialDepDefinition, Type[T]]:
138
    """
139
    Takes a function or a generator and returns a function and the return type.
140
    :param func:
141
    :return:
142
    """
143
144 1
    return_type = reflect(func).return_type
145 1
    if not return_type:
146 1
        raise MissingReturnType(
147
            f"Function {func.__name__} used as a definition must have a return type"
148
        )
149
150 1
    if not inspect.isgeneratorfunction(func) and not inspect.isasyncgenfunction(func):
151 1
        return construction(func), return_type
152
153 1
    return (
154
        yielding_construction(func),
155
        _generator_type(return_type),
156
    )
157
158
159 1
def _generator_type(return_type):
160
    return return_type.__args__[0]  # todo: something less hacky
161