aiohttp_rpc.middlewares.BaseJsonRpcMiddleware.__init__() - Code Metrics - Inspection of "Update python version" - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 3314b0...3c70d5 )

by Michael

created 2020-05-26 14:35 UTC

BaseJsonRpcMiddleware.init() A

↳ Parent: aiohttp_rpc.middlewares

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	3
dl	0
loc	3
rs	10
c	0
b	0
f	0

import abc
import logging
import typing

from . import errors, protocol


if typing.TYPE_CHECKING:
    from . import server as rpc_server

__all__ = (
    'BaseJsonRpcMiddleware',
    'ExceptionMiddleware',
    'DEFAULT_MIDDLEWARES',
)


class BaseJsonRpcMiddleware(abc.ABC):
    server: 'rpc_server.JsonRpcServer'
    get_response: typing.Callable

    def __init__(self, server: 'rpc_server.JsonRpcServer', get_response: typing.Callable) -> None:
        self.server = server
        self.get_response = get_response

    @abc.abstractmethod
    async def __call__(self, request: protocol.JsonRpcRequest) -> protocol.JsonRpcResponse:
        # Code to be executed for each request before the method are called.

        response = await self.get_response(request)

        # Code to be executed for each request/response after the method is called.

        return response


class ExceptionMiddleware(BaseJsonRpcMiddleware):
    async def __call__(self, request: protocol.JsonRpcRequest) -> protocol.JsonRpcResponse:
        try:
            response = await self.get_response(request)
        except Exception as e:
            logging.exception(e)
            response = protocol.JsonRpcResponse(
                msg_id=request.msg_id,
                jsonrpc=request.jsonrpc,
                error=errors.InternalError().with_traceback(),
            )

        return response


DEFAULT_MIDDLEWARES = (
    ExceptionMiddleware,
)


1			import abc
2			import logging
3			import typing
4
5			from . import errors, protocol
6
7
8			if typing.TYPE_CHECKING:
9			from . import server as rpc_server
10
11			__all__ = (
12			'BaseJsonRpcMiddleware',
13			'ExceptionMiddleware',
14			'DEFAULT_MIDDLEWARES',
15			)
16
17
18			class BaseJsonRpcMiddleware(abc.ABC):
19			server: 'rpc_server.JsonRpcServer'
20			get_response: typing.Callable
21
22			def __init__(self, server: 'rpc_server.JsonRpcServer', get_response: typing.Callable) -> None:
23			self.server = server
24			self.get_response = get_response
25
26			@abc.abstractmethod
27			async def __call__(self, request: protocol.JsonRpcRequest) -> protocol.JsonRpcResponse:
28			# Code to be executed for each request before the method are called.
29
30			response = await self.get_response(request)
31
32			# Code to be executed for each request/response after the method is called.
33
34			return response
35
36
37			class ExceptionMiddleware(BaseJsonRpcMiddleware):
38			async def __call__(self, request: protocol.JsonRpcRequest) -> protocol.JsonRpcResponse:
39			try:
40			response = await self.get_response(request)
41			except Exception as e:
42			logging.exception(e)
43			response = protocol.JsonRpcResponse(
44			msg_id=request.msg_id,
45			jsonrpc=request.jsonrpc,
46			error=errors.InternalError().with_traceback(),
47			)
48
49			return response
50
51
52			DEFAULT_MIDDLEWARES = (
53			ExceptionMiddleware,
54			)
55

expert-m / aiohttp-rpc

Push — master ( 3314b0...3c70d5 )

BaseJsonRpcMiddleware.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

BaseJsonRpcMiddleware.init() A