aiohttp_rpc.utils.parse_args_and_kwargs() - Code Metrics - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

aiohttp_rpc.utils.parse_args_and_kwargs() B
last analyzed 2023-10-15 08:14 UTC

↳ Parent: aiohttp_rpc.utils

Complexity

Conditions

Size

Total Lines	19
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	19
rs	8
c	0
b	0
f	0
cc	7
nop	2

import json
import typing
import uuid
from functools import partial
from traceback import format_exception_only

from . import constants, errors

if typing.TYPE_CHECKING:
    from . import protocol  # NOQA


__all__ = (
    'convert_params_to_args_and_kwargs',
    'parse_args_and_kwargs',
    'get_exc_message',
    'json_serialize',
    'collect_batch_result',
)


def convert_params_to_args_and_kwargs(params: typing.Any) -> typing.Tuple[typing.Sequence, typing.Mapping]:
    if params is constants.NOTHING:
        return (), {}

    if isinstance(params, constants.JSON_PRIMITIVE_TYPES):
        return (params,), {}

    if isinstance(params, typing.Sequence):
        return params, {}

    if isinstance(params, typing.Mapping):
        return (), params

    raise errors.InvalidParams('Params have unsupported data types.')


def parse_args_and_kwargs(args: typing.Optional[typing.Sequence],
                          kwargs: typing.Optional[typing.Mapping],
                          ) -> typing.Tuple[typing.Any, typing.Sequence, typing.Mapping]:
    has_args = bool(args)
    has_kwargs = bool(kwargs)

    if not has_args and not has_kwargs:
        return constants.NOTHING, (), {}  # type: ignore

    if not (has_args ^ has_kwargs):
        raise errors.InvalidParams('Need use args or kwargs.')

    if has_args:
        if len(args) == 1 and isinstance(args[0], constants.JSON_PRIMITIVE_TYPES):  # type: ignore
            return args[0], args, {}  # type: ignore

        return args, args, {}  # type: ignore

    return kwargs, (), kwargs  # type: ignore


def get_random_id() -> str:
    return str(uuid.uuid4())


def get_exc_message(exp: BaseException) -> str:
    return ''.join(format_exception_only(exp.__class__, exp)).strip()


def validate_jsonrpc(jsonrpc: typing.Any) -> None:
    if jsonrpc != constants.VERSION_2_0:
        raise errors.InvalidRequest(f'Only version "{constants.VERSION_2_0}" is supported.')


def collect_batch_result(batch_request: 'protocol.JsonRpcBatchRequest',
                         batch_response: 'protocol.JsonRpcBatchResponse') -> typing.Tuple[typing.Any, ...]:
    from . import protocol

    unlinked_results = protocol.JsonRpcUnlinkedResults()
    responses_map: typing.Dict[typing.Any, typing.Any] = {}

    for response in batch_response.responses:
        if response.error is None:
            value = response.result
        else:
            value = response.error

        if response.id is None:
            unlinked_results.add(value)
            continue

        if response.id in responses_map:
            if isinstance(responses_map[response.id], protocol.JsonRpcDuplicatedResults):

                responses_map[response.id].add(value)
            else:
                responses_map[response.id] = protocol.JsonRpcDuplicatedResults([
                    responses_map[response.id],
                    value,
                ])
        else:
            responses_map[response.id] = value

    return tuple(
        unlinked_results or None
        if request.is_notification
        else responses_map.get(request.id, unlinked_results or None)
        for request in batch_request.requests
    )


json_serialize = partial(json.dumps, default=lambda x: repr(x))
json_deserialize = json.loads


1			import json
2			import typing
3			import uuid
4			from functools import partial
5			from traceback import format_exception_only
6
7			from . import constants, errors
8
9			if typing.TYPE_CHECKING:
10			from . import protocol # NOQA
11
12
13			__all__ = (
14			'convert_params_to_args_and_kwargs',
15			'parse_args_and_kwargs',
16			'get_exc_message',
17			'json_serialize',
18			'collect_batch_result',
19			)
20
21
22			def convert_params_to_args_and_kwargs(params: typing.Any) -> typing.Tuple[typing.Sequence, typing.Mapping]:
23			if params is constants.NOTHING:
24			return (), {}
25
26			if isinstance(params, constants.JSON_PRIMITIVE_TYPES):
27			return (params,), {}
28
29			if isinstance(params, typing.Sequence):
30			return params, {}
31
32			if isinstance(params, typing.Mapping):
33			return (), params
34
35			raise errors.InvalidParams('Params have unsupported data types.')
36
37
38			def parse_args_and_kwargs(args: typing.Optional[typing.Sequence],
39			kwargs: typing.Optional[typing.Mapping],
40			) -> typing.Tuple[typing.Any, typing.Sequence, typing.Mapping]:
41			has_args = bool(args)
42			has_kwargs = bool(kwargs)
43
44			if not has_args and not has_kwargs:
45			return constants.NOTHING, (), {} # type: ignore
46
47			if not (has_args ^ has_kwargs):
48			raise errors.InvalidParams('Need use args or kwargs.')
49
50			if has_args:
51			if len(args) == 1 and isinstance(args[0], constants.JSON_PRIMITIVE_TYPES): # type: ignore
52			return args[0], args, {} # type: ignore
53
54			return args, args, {} # type: ignore
55
56			return kwargs, (), kwargs # type: ignore
57
58
59			def get_random_id() -> str:
60			return str(uuid.uuid4())
61
62
63			def get_exc_message(exp: BaseException) -> str:
64			return ''.join(format_exception_only(exp.__class__, exp)).strip()
65
66
67			def validate_jsonrpc(jsonrpc: typing.Any) -> None:
68			if jsonrpc != constants.VERSION_2_0:
69			raise errors.InvalidRequest(f'Only version "{constants.VERSION_2_0}" is supported.')
70
71
72			def collect_batch_result(batch_request: 'protocol.JsonRpcBatchRequest',
73			batch_response: 'protocol.JsonRpcBatchResponse') -> typing.Tuple[typing.Any, ...]:
74			from . import protocol
75
76			unlinked_results = protocol.JsonRpcUnlinkedResults()
77			responses_map: typing.Dict[typing.Any, typing.Any] = {}
78
79			for response in batch_response.responses:
80			if response.error is None:
81			value = response.result
82			else:
83			value = response.error
84
85			if response.id is None:
86			unlinked_results.add(value)
87			continue
88
89			if response.id in responses_map:
90			if isinstance(responses_map[response.id], protocol.JsonRpcDuplicatedResults):
			0 ignored issues – show introduced 2021-08-02 01:59 UTC by Report Bug Copy Issue Report The variable `protocol` does not seem to be defined in case `typing.TYPE_CHECKING` on line `9` is `False`. Are you sure this can never be the case? Loading history...
91			responses_map[response.id].add(value)
92			else:
93			responses_map[response.id] = protocol.JsonRpcDuplicatedResults([
94			responses_map[response.id],
95			value,
96			])
97			else:
98			responses_map[response.id] = value
99
100			return tuple(
101			unlinked_results or None
102			if request.is_notification
103			else responses_map.get(request.id, unlinked_results or None)
104			for request in batch_request.requests
105			)
106
107
108			json_serialize = partial(json.dumps, default=lambda x: repr(x))
109			json_deserialize = json.loads
110

expert-m / aiohttp-rpc

aiohttp_rpc.utils.parse_args_and_kwargs() B last analyzed 2023-10-15 08:14 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

aiohttp_rpc.utils.parse_args_and_kwargs() B
last analyzed 2023-10-15 08:14 UTC