jsons.decorators._get_wrapper() - Code Metrics - Inspection of "Splitted functions into several." - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5a99d0...988c2c )

by Ramon

created 2019-03-17 12:07 UTC

jsons.decorators._get_wrapper() A

↳ Parent: jsons.decorators

Complexity

Conditions

Size

Total Lines	16
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	16
rs	9.65
c	0
b	0
f	0
cc	3
nop	6

"""
This module contains decorators that facilitate the `jsons` functions in an
alternative fashion.
"""
from inspect import signature, Parameter, isawaitable, iscoroutinefunction
from jsons import JsonSerializable, dump, load, loads, loadb, dumps, dumpb
from jsons.exceptions import InvalidDecorationError


def loaded(
        parameters=True,
        returnvalue=True,
        fork_inst=JsonSerializable,
        loader=load,
        **kwargs):
    """
    Return a decorator that can call `jsons.load` upon all parameters and the
    return value of the decorated function.


    **Example**:

    >>> from datetime import datetime
    >>> @loaded()
    ... def func(arg: datetime) -> datetime:
    ...     # arg is now of type datetime.
    ...     return '2018-10-04T21:57:00Z'  # This will become a datetime.
    >>> res = func('2018-10-04T21:57:00Z')
    >>> type(res).__name__
    'datetime'

    :param parameters: determines whether parameters should be taken into
    account.
    :param returnvalue: determines whether the return value should be taken
    into account.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :param kwargs: any keyword arguments that should be passed on to
    `jsons.load`
    :param loader: the load function which must be one of (``load``,
    ``loads``, ``loadb``)
    :return: a decorator that can be placed on a function.
    """
    if loader not in (load, loads, loadb):
        raise InvalidDecorationError("The 'loader' argument must be one of: "
                                     "jsons.load, jsons.loads, jsons.loadb")
    return _get_decorator(parameters, returnvalue, fork_inst, loader, kwargs)


def dumped(
        parameters=True,
        returnvalue=True,
        fork_inst=JsonSerializable,
        dumper=dump,
        **kwargs):
    """
    Return a decorator that can call `jsons.dump` upon all parameters and the
    return value of the decorated function.


    **Example**:

    >>> from datetime import datetime
    >>> @dumped()
    ... def func(arg):
    ...     # arg is now of type str.
    ...     return datetime.now()
    >>> res = func(datetime.now())
    >>> type(res).__name__
    'str'

    :param parameters: determines whether parameters should be taken into
    account.
    :param returnvalue: determines whether the return value should be taken
    into account.
    :param fork_inst: if given, it uses this fork of ``JsonSerializable``.
    :param kwargs: any keyword arguments that should be passed on to
    `jsons.dump`
    :param dumper: the dump function which must be one of (``dump``,
    ``dumps``, ``dumpb``)
    :return: a decorator that can be placed on a function.
    """
    if dumper not in (dump, dumps, dumpb):
        raise InvalidDecorationError("The 'dumper' argument must be one of: "
                                     "jsons.dump, jsons.dumps, jsons.dumpb")
    return _get_decorator(parameters, returnvalue, fork_inst, dumper, kwargs)


def _get_decorator(parameters, returnvalue, fork_inst, mapper, mapper_kwargs):
    def _decorator(decorated):
        _validate_decoration(decorated, fork_inst)
        args = [decorated, parameters, returnvalue,
                fork_inst, mapper, mapper_kwargs]
        wrapper = (_get_async_wrapper(*args) if iscoroutinefunction(decorated)
                   else _get_wrapper(*args))
        return wrapper
    return _decorator


def _get_wrapper(
        decorated,
        parameters,
        returnvalue,
        fork_inst,
        mapper,
        mapper_kwargs):
    def _wrapper(*args, **kwargs):
        result = _run_decorated(decorated, mapper if parameters else None,
                                fork_inst, args, kwargs, mapper_kwargs)
        if returnvalue:
            result = _map_returnvalue(result, decorated, fork_inst, mapper,
                                      mapper_kwargs)
        return result

    return _wrapper


def _get_async_wrapper(
        decorated,
        parameters,
        returnvalue,
        fork_inst,
        mapper,
        mapper_kwargs):

    async def _async_wrapper(*args, **kwargs):
        result = _run_decorated(decorated, mapper if parameters else None,
                                fork_inst, args, kwargs, mapper_kwargs)
        if isawaitable(result):
            result = await result
        if returnvalue:
            result = _map_returnvalue(result, decorated, fork_inst, mapper,
                                      mapper_kwargs)
        return result

    return _async_wrapper


def _get_params_sig(args, func):
    sig = signature(func)
    params = sig.parameters
    param_names = [param_name for param_name in params]
    result = [(args[i], params[param_names[i]]) for i in range(len(args))]
    return result


def _map_args(args, decorated, fork_inst, mapper, mapper_kwargs):
    params_sig = _get_params_sig(args, decorated)
    new_args = []
    for arg, sig in params_sig:
        if sig.name in ('self', 'cls') and hasattr(arg, decorated.__name__):
            # `decorated` is a method and arg is either `self` or `cls`.
            new_arg = arg
        else:
            cls = sig.annotation if sig.annotation != Parameter.empty else None
            new_arg = mapper(arg, cls=cls, fork_inst=fork_inst,
                             **mapper_kwargs)

        new_args.append(new_arg)
    return new_args


def _map_returnvalue(returnvalue, decorated, fork_inst, mapper, mapper_kwargs):
    return_annotation = signature(decorated).return_annotation
    cls = return_annotation if return_annotation != Parameter.empty else None
    result = mapper(returnvalue, cls=cls, fork_inst=fork_inst, **mapper_kwargs)
    return result


def _run_decorated(decorated, mapper, fork_inst, args, kwargs, mapper_kwargs):
    new_args = args
    if mapper:
        new_args = _map_args(args, decorated, fork_inst, mapper, mapper_kwargs)
    result = decorated(*new_args, **kwargs)
    return result


def _validate_decoration(decorated, fork_inst):
    if isinstance(decorated, staticmethod):
        fork_inst._warn('You cannot decorate a static- or classmethod. '
                        'You can still obtain the desired behavior by '
                        'decorating your method first and then place '
                        '@staticmethod/@classmethod on top (switching the '
                        'order).')
        raise InvalidDecorationError(
            'Cannot decorate a static- or classmethod.')
    if isinstance(decorated, type):
        raise InvalidDecorationError('Cannot decorate a class.')


1			"""
2			This module contains decorators that facilitate the `jsons` functions in an
3			alternative fashion.
4			"""
5			from inspect import signature, Parameter, isawaitable, iscoroutinefunction
6			from jsons import JsonSerializable, dump, load, loads, loadb, dumps, dumpb
7			from jsons.exceptions import InvalidDecorationError
8
9
10			def loaded(
11			parameters=True,
12			returnvalue=True,
13			fork_inst=JsonSerializable,
14			loader=load,
15			**kwargs):
16			"""
17			Return a decorator that can call `jsons.load` upon all parameters and the
18			return value of the decorated function.
19
20
21			Example:
22
23			>>> from datetime import datetime
24			>>> @loaded()
25			... def func(arg: datetime) -> datetime:
26			... # arg is now of type datetime.
27			... return '2018-10-04T21:57:00Z' # This will become a datetime.
28			>>> res = func('2018-10-04T21:57:00Z')
29			>>> type(res).__name__
30			'datetime'
31
32			:param parameters: determines whether parameters should be taken into
33			account.
34			:param returnvalue: determines whether the return value should be taken
35			into account.
36			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
37			:param kwargs: any keyword arguments that should be passed on to
38			`jsons.load`
39			:param loader: the load function which must be one of (``load``,
40			``loads``, ``loadb``)
41			:return: a decorator that can be placed on a function.
42			"""
43			if loader not in (load, loads, loadb):
44			raise InvalidDecorationError("The 'loader' argument must be one of: "
45			"jsons.load, jsons.loads, jsons.loadb")
46			return _get_decorator(parameters, returnvalue, fork_inst, loader, kwargs)
47
48
49			def dumped(
50			parameters=True,
51			returnvalue=True,
52			fork_inst=JsonSerializable,
53			dumper=dump,
54			**kwargs):
55			"""
56			Return a decorator that can call `jsons.dump` upon all parameters and the
57			return value of the decorated function.
58
59
60			Example:
61
62			>>> from datetime import datetime
63			>>> @dumped()
64			... def func(arg):
65			... # arg is now of type str.
66			... return datetime.now()
67			>>> res = func(datetime.now())
68			>>> type(res).__name__
69			'str'
70
71			:param parameters: determines whether parameters should be taken into
72			account.
73			:param returnvalue: determines whether the return value should be taken
74			into account.
75			:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
76			:param kwargs: any keyword arguments that should be passed on to
77			`jsons.dump`
78			:param dumper: the dump function which must be one of (``dump``,
79			``dumps``, ``dumpb``)
80			:return: a decorator that can be placed on a function.
81			"""
82			if dumper not in (dump, dumps, dumpb):
83			raise InvalidDecorationError("The 'dumper' argument must be one of: "
84			"jsons.dump, jsons.dumps, jsons.dumpb")
85			return _get_decorator(parameters, returnvalue, fork_inst, dumper, kwargs)
86
87
88			def _get_decorator(parameters, returnvalue, fork_inst, mapper, mapper_kwargs):
89			def _decorator(decorated):
90			_validate_decoration(decorated, fork_inst)
91			args = [decorated, parameters, returnvalue,
92			fork_inst, mapper, mapper_kwargs]
93			wrapper = (_get_async_wrapper(*args) if iscoroutinefunction(decorated)
94			else _get_wrapper(*args))
95			return wrapper
96			return _decorator
97
98
99			def _get_wrapper(
100			decorated,
101			parameters,
102			returnvalue,
103			fork_inst,
104			mapper,
105			mapper_kwargs):
106			def _wrapper(args, *kwargs):
107			result = _run_decorated(decorated, mapper if parameters else None,
108			fork_inst, args, kwargs, mapper_kwargs)
109			if returnvalue:
110			result = _map_returnvalue(result, decorated, fork_inst, mapper,
111			mapper_kwargs)
112			return result
113
114			return _wrapper
115
116
117			def _get_async_wrapper(
118			decorated,
119			parameters,
120			returnvalue,
121			fork_inst,
122			mapper,
123			mapper_kwargs):
124
125			async def _async_wrapper(args, *kwargs):
126			result = _run_decorated(decorated, mapper if parameters else None,
127			fork_inst, args, kwargs, mapper_kwargs)
128			if isawaitable(result):
129			result = await result
130			if returnvalue:
131			result = _map_returnvalue(result, decorated, fork_inst, mapper,
132			mapper_kwargs)
133			return result
134
135			return _async_wrapper
136
137
138			def _get_params_sig(args, func):
139			sig = signature(func)
140			params = sig.parameters
141			param_names = [param_name for param_name in params]
142			result = [(args[i], params[param_names[i]]) for i in range(len(args))]
143			return result
144
145
146			def _map_args(args, decorated, fork_inst, mapper, mapper_kwargs):
147			params_sig = _get_params_sig(args, decorated)
148			new_args = []
149			for arg, sig in params_sig:
150			if sig.name in ('self', 'cls') and hasattr(arg, decorated.__name__):
151			# `decorated` is a method and arg is either `self` or `cls`.
152			new_arg = arg
153			else:
154			cls = sig.annotation if sig.annotation != Parameter.empty else None
155			new_arg = mapper(arg, cls=cls, fork_inst=fork_inst,
156			**mapper_kwargs)
157
158			new_args.append(new_arg)
159			return new_args
160
161
162			def _map_returnvalue(returnvalue, decorated, fork_inst, mapper, mapper_kwargs):
163			return_annotation = signature(decorated).return_annotation
164			cls = return_annotation if return_annotation != Parameter.empty else None
165			result = mapper(returnvalue, cls=cls, fork_inst=fork_inst, **mapper_kwargs)
166			return result
167
168
169			def _run_decorated(decorated, mapper, fork_inst, args, kwargs, mapper_kwargs):
170			new_args = args
171			if mapper:
172			new_args = _map_args(args, decorated, fork_inst, mapper, mapper_kwargs)
173			result = decorated(new_args, *kwargs)
174			return result
175
176
177			def _validate_decoration(decorated, fork_inst):
178			if isinstance(decorated, staticmethod):
179			fork_inst._warn('You cannot decorate a static- or classmethod. '
180			'You can still obtain the desired behavior by '
181			'decorating your method first and then place '
182			'@staticmethod/@classmethod on top (switching the '
183			'order).')
184			raise InvalidDecorationError(
185			'Cannot decorate a static- or classmethod.')
186			if isinstance(decorated, type):
187			raise InvalidDecorationError('Cannot decorate a class.')
188

ramonhagenaars / jsons

Push — master ( 5a99d0...988c2c )

jsons.decorators._get_wrapper() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like