lagom.wrapping   A
last analyzed

Complexity

Total Complexity 5

Size/Duplication

Total Lines 57
Duplicated Lines 0 %

Test Coverage

Coverage 100%

Importance

Changes 0
Metric Value
wmc 5
eloc 30
dl 0
loc 57
ccs 25
cts 25
cp 1
rs 10
c 0
b 0
f 0

2 Functions

Rating   Name   Duplication   Size   Complexity  
A _wrap_func_in_error_handling() 0 23 2
A apply_argument_updater() 0 19 3
1
"""
2
Code in this module is used to wrap and decorate functions that have been
3
bound to a container
4
"""
5
6 1
import functools
7 1
import inspect
8
9 1
from .exceptions import UnableToInvokeBoundFunction
10 1
from .util.reflection import FunctionSpec
11
12
13 1
def apply_argument_updater(
14
    func, argument_updater, spec: FunctionSpec, catch_errors=False
15
):
16 1
    inner_func = func if not catch_errors else _wrap_func_in_error_handling(func, spec)
17 1
    if inspect.iscoroutinefunction(func):
18
19 1
        @functools.wraps(func)
20 1
        async def _bound_func(*args, **kwargs):
21 1
            bound_args, bound_kwargs = argument_updater(args, kwargs)
22 1
            return await inner_func(*bound_args, **bound_kwargs)
23
24
    else:
25
26 1
        @functools.wraps(func)
27 1
        def _bound_func(*args, **kwargs):
28 1
            bound_args, bound_kwargs = argument_updater(args, kwargs)
29 1
            return inner_func(*bound_args, **bound_kwargs)
30
31 1
    return _bound_func
32
33
34 1
def _wrap_func_in_error_handling(func, spec: FunctionSpec):
35
    """
36
    Takes a func and its spec and returns a function that's the same
37
    but with more useful TypeError messages
38
    :param func:
39
    :param spec:
40
    :return:
41
    """
42
43 1
    @functools.wraps(func)
44 1
    def _error_handling_func(*args, **kwargs):
45 1
        try:
46 1
            return func(*args, **kwargs)
47 1
        except TypeError as error:
48
            # if it wasn't in kwargs the container couldn't build it
49 1
            unresolvable_deps = [
50
                dep_type
51
                for (name, dep_type) in spec.annotations.items()
52
                if name not in kwargs.keys()
53
            ]
54 1
            raise UnableToInvokeBoundFunction(str(error), unresolvable_deps)
55
56
    return _error_handling_func
57