jsons.decorators._get_async_wrapper() - Code Metrics - ramonhagenaars/jsons - Measure and Improve Code Quality continuously with Scrutinizer

jsons.decorators._get_async_wrapper() A
last analyzed 2022-06-09 19:52 UTC

↳ Parent: jsons.decorators

Complexity

Conditions

Size

Total Lines	18
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

ramonhagenaars / jsons

jsons.decorators._get_async_wrapper() A last analyzed 2022-06-09 19:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

jsons.decorators._get_async_wrapper() A
last analyzed 2022-06-09 19:52 UTC