aiohttp_rpc.server.base.BaseJsonRpcServer.load_middlewares() - Code Metrics - Inspection of "Update method name & change some typings" - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 7dbe03...d2ae7d )

by Michael

created 2021-08-01 07:05 UTC

ares() A

↳ Parent: aiohttp_rpc.server.base.BaseJsonRpcServer.add_method()

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	7
rs	10
c	0
b	0
f	0
cc	2
nop	1

import abc
import asyncio
import typing
from functools import partial

from .. import errors, protocol, typedefs, utils


__all__ = (
    'BaseJsonRpcServer',
)


class BaseJsonRpcServer(abc.ABC):
    methods: typing.MutableMapping[str, protocol.BaseJsonRpcMethod]
    middlewares: typing.Sequence[typing.Callable]
    json_serialize: typedefs.UnboundJSONEncoderType
    _middleware_chain: typedefs.UnboundSingleRequestProcessorType

    def __init__(self, *,
                 json_serialize: typedefs.JSONEncoderType = utils.json_serialize,
                 middlewares: typing.Sequence = (),
                 methods: typing.Optional[typing.Dict[str, protocol.BaseJsonRpcMethod]] = None) -> None:
        if methods is None:
            methods = {
                'get_method': protocol.JsonRpcMethod('', self.get_method),
                'get_methods': protocol.JsonRpcMethod('', self.get_methods),
            }

        self.methods = methods

        self.middlewares = middlewares
        self._load_middlewares()

        self.json_serialize = json_serialize  # type: ignore

    def add_method(self,
                   method: typing.Union[typedefs.ServerMethodDescriptionType], *,
                   replace: bool = False) -> protocol.BaseJsonRpcMethod:
        if not isinstance(method, protocol.BaseJsonRpcMethod):
            if callable(method):
                method = protocol.JsonRpcMethod('', method)
            else:
                method = protocol.JsonRpcMethod(*method)

        if not replace and method.name in self.methods:
            raise errors.InvalidParams(f'Method {method.name} has already been added.')

        self.methods[method.name] = method

        return method

    def add_methods(self,
                    methods: typing.Iterable[typedefs.ServerMethodDescriptionType], *,
                    replace: bool = False) -> typing.Tuple[protocol.BaseJsonRpcMethod, ...]:
        return tuple(
            self.add_method(method, replace=replace)
            for method in methods
        )

    async def call(self,
                   method_name: str, *,
                   args: typing.Optional[typing.Sequence] = None,
                   kwargs: typing.Optional[typing.Mapping] = None,
                   extra_args: typing.Optional[typing.Mapping] = None) -> typing.Any:
        if args is None:
            args = ()

        if kwargs is None:
            kwargs = {}

        if method_name not in self.methods:
            raise errors.MethodNotFound

        return await self.methods[method_name](args=args, kwargs=kwargs, extra_args=extra_args)

    def get_methods(self) -> typing.Mapping[str, typing.Mapping[str, typing.Any]]:
        return {
            name: {
                'doc': method.func.__doc__,
                'args': method.supported_args,
                'kwargs': method.supported_kwargs,
            }
            for name, method in self.methods.items()
        }

    def get_method(self, name: str) -> typing.Optional[typing.Mapping[str, typing.Any]]:
        method = self.methods.get(name)

        if not method:
            return None

        return {
            'doc': method.func.__doc__,
            'args': method.supported_args,
            'kwargs': method.supported_kwargs,
        }

    def _load_middlewares(self) -> None:
        self._middleware_chain = self._process_single_request  # type: ignore

        for middleware in reversed(self.middlewares):
            self._middleware_chain: typedefs.SingleRequestProcessorType = partial(  # type: ignore
                middleware,
                handler=self._middleware_chain,
            )

    async def _process_input_data(
            self,
            data: typing.Any, *,
            context: typing.MutableMapping[str, typing.Any],
    ) -> typing.Optional[typing.Union[typing.Mapping, typing.Tuple[typing.Mapping, ...]]]:
        if isinstance(data, typing.Sequence):
            if not data:
                return protocol.JsonRpcResponse(error=errors.InvalidRequest()).dump()

            json_responses = await asyncio.gather(
                *(
                    self._process_single_json_request(raw_rcp_request, context=context)

                    for raw_rcp_request in data
                ),
                return_exceptions=True,
            )

            result = tuple(
                json_response
                for json_response in self._prepare_exceptions(json_responses)
                if json_response is not None
            )

            return result if result else None

        if isinstance(data, typing.Mapping):
            return await self._process_single_json_request(data, context=context)

        response = protocol.JsonRpcResponse(error=errors.InvalidRequest('Data must be a dict or an list.'))
        return response.dump()

    @staticmethod
    def _prepare_exceptions(values: typing.Iterable) -> typing.Iterable:
        for i, value in enumerate(values):
            if isinstance(value, errors.JsonRpcError):
                yield protocol.JsonRpcResponse(error=value)
            elif isinstance(value, Exception):
                raise value
            else:
                yield value

    async def _process_single_json_request(self,
                                           json_request: typing.Any, *,
                                           context: typing.MutableMapping[str, typing.Any],
                                           ) -> typing.Optional[typing.Mapping]:
        if not isinstance(json_request, typing.Mapping):
            return protocol.JsonRpcResponse(error=errors.InvalidRequest('Data must be a dict.')).dump()

        try:
            request = protocol.JsonRpcRequest.load(json_request, context=context)
        except errors.JsonRpcError as e:
            return protocol.JsonRpcResponse(id=json_request.get('id'), error=e).dump()

        response = await self._middleware_chain(request)

        if response.is_notification:
            return None

        return response.dump()

    async def _process_single_request(self, request: protocol.JsonRpcRequest) -> protocol.JsonRpcResponse:
        result, error = None, None

        try:
            result = await self.call(
                request.method_name,
                args=request.args,
                kwargs=request.kwargs,
                extra_args=request.extra_args,
            )
        except errors.JsonRpcError as e:
            error = e

        response = protocol.JsonRpcResponse(
            id=request.id,
            jsonrpc=request.jsonrpc,
            result=result,
            error=error,
        )

        return response


1			import abc
2			import asyncio
3			import typing
4			from functools import partial
5
6			from .. import errors, protocol, typedefs, utils
7
8
9			__all__ = (
10			'BaseJsonRpcServer',
11			)
12
13
14			class BaseJsonRpcServer(abc.ABC):
15			methods: typing.MutableMapping[str, protocol.BaseJsonRpcMethod]
16			middlewares: typing.Sequence[typing.Callable]
17			json_serialize: typedefs.UnboundJSONEncoderType
18			_middleware_chain: typedefs.UnboundSingleRequestProcessorType
19
20			def __init__(self, *,
21			json_serialize: typedefs.JSONEncoderType = utils.json_serialize,
22			middlewares: typing.Sequence = (),
23			methods: typing.Optional[typing.Dict[str, protocol.BaseJsonRpcMethod]] = None) -> None:
24			if methods is None:
25			methods = {
26			'get_method': protocol.JsonRpcMethod('', self.get_method),
27			'get_methods': protocol.JsonRpcMethod('', self.get_methods),
28			}
29
30			self.methods = methods
31
32			self.middlewares = middlewares
33			self._load_middlewares()
34
35			self.json_serialize = json_serialize # type: ignore
36
37			def add_method(self,
38			method: typing.Union[typedefs.ServerMethodDescriptionType], *,
39			replace: bool = False) -> protocol.BaseJsonRpcMethod:
40			if not isinstance(method, protocol.BaseJsonRpcMethod):
41			if callable(method):
42			method = protocol.JsonRpcMethod('', method)
43			else:
44			method = protocol.JsonRpcMethod(*method)
45
46			if not replace and method.name in self.methods:
47			raise errors.InvalidParams(f'Method {method.name} has already been added.')
48
49			self.methods[method.name] = method
50
51			return method
52
53			def add_methods(self,
54			methods: typing.Iterable[typedefs.ServerMethodDescriptionType], *,
55			replace: bool = False) -> typing.Tuple[protocol.BaseJsonRpcMethod, ...]:
56			return tuple(
57			self.add_method(method, replace=replace)
58			for method in methods
59			)
60
61			async def call(self,
62			method_name: str, *,
63			args: typing.Optional[typing.Sequence] = None,
64			kwargs: typing.Optional[typing.Mapping] = None,
65			extra_args: typing.Optional[typing.Mapping] = None) -> typing.Any:
66			if args is None:
67			args = ()
68
69			if kwargs is None:
70			kwargs = {}
71
72			if method_name not in self.methods:
73			raise errors.MethodNotFound
74
75			return await self.methods[method_name](args=args, kwargs=kwargs, extra_args=extra_args)
76
77			def get_methods(self) -> typing.Mapping[str, typing.Mapping[str, typing.Any]]:
78			return {
79			name: {
80			'doc': method.func.__doc__,
81			'args': method.supported_args,
82			'kwargs': method.supported_kwargs,
83			}
84			for name, method in self.methods.items()
85			}
86
87			def get_method(self, name: str) -> typing.Optional[typing.Mapping[str, typing.Any]]:
88			method = self.methods.get(name)
89
90			if not method:
91			return None
92
93			return {
94			'doc': method.func.__doc__,
95			'args': method.supported_args,
96			'kwargs': method.supported_kwargs,
97			}
98
99			def _load_middlewares(self) -> None:
100			self._middleware_chain = self._process_single_request # type: ignore
101
102			for middleware in reversed(self.middlewares):
103			self._middleware_chain: typedefs.SingleRequestProcessorType = partial( # type: ignore
104			middleware,
105			handler=self._middleware_chain,
106			)
107
108			async def _process_input_data(
109			self,
110			data: typing.Any, *,
111			context: typing.MutableMapping[str, typing.Any],
112			) -> typing.Optional[typing.Union[typing.Mapping, typing.Tuple[typing.Mapping, ...]]]:
113			if isinstance(data, typing.Sequence):
114			if not data:
115			return protocol.JsonRpcResponse(error=errors.InvalidRequest()).dump()
116
117			json_responses = await asyncio.gather(
118			*(
119			self._process_single_json_request(raw_rcp_request, context=context)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-04 05:01 UTC by Report Bug Copy Issue Report The variable `raw_rcp_request` does not seem to be defined. Loading history...
120			for raw_rcp_request in data
121			),
122			return_exceptions=True,
123			)
124
125			result = tuple(
126			json_response
127			for json_response in self._prepare_exceptions(json_responses)
128			if json_response is not None
129			)
130
131			return result if result else None
132
133			if isinstance(data, typing.Mapping):
134			return await self._process_single_json_request(data, context=context)
135
136			response = protocol.JsonRpcResponse(error=errors.InvalidRequest('Data must be a dict or an list.'))
137			return response.dump()
138
139			@staticmethod
140			def _prepare_exceptions(values: typing.Iterable) -> typing.Iterable:
141			for i, value in enumerate(values):
142			if isinstance(value, errors.JsonRpcError):
143			yield protocol.JsonRpcResponse(error=value)
144			elif isinstance(value, Exception):
145			raise value
146			else:
147			yield value
148
149			async def _process_single_json_request(self,
150			json_request: typing.Any, *,
151			context: typing.MutableMapping[str, typing.Any],
152			) -> typing.Optional[typing.Mapping]:
153			if not isinstance(json_request, typing.Mapping):
154			return protocol.JsonRpcResponse(error=errors.InvalidRequest('Data must be a dict.')).dump()
155
156			try:
157			request = protocol.JsonRpcRequest.load(json_request, context=context)
158			except errors.JsonRpcError as e:
159			return protocol.JsonRpcResponse(id=json_request.get('id'), error=e).dump()
160
161			response = await self._middleware_chain(request)
162
163			if response.is_notification:
164			return None
165
166			return response.dump()
167
168			async def _process_single_request(self, request: protocol.JsonRpcRequest) -> protocol.JsonRpcResponse:
169			result, error = None, None
170
171			try:
172			result = await self.call(
173			request.method_name,
174			args=request.args,
175			kwargs=request.kwargs,
176			extra_args=request.extra_args,
177			)
178			except errors.JsonRpcError as e:
179			error = e
180
181			response = protocol.JsonRpcResponse(
182			id=request.id,
183			jsonrpc=request.jsonrpc,
184			result=result,
185			error=error,
186			)
187
188			return response
189

expert-m / aiohttp-rpc

Push — master ( 7dbe03...d2ae7d )

ares() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like