aiohttp_rpc.server.http.JsonRpcServer.handle_http_request() - Code Metrics - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

JsonRpcServer.handle_http_request() A
last analyzed 2023-10-15 08:14 UTC

↳ Parent: aiohttp_rpc.server.http

Complexity

Conditions

Size

Total Lines	13
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	13
rs	9.9
c	0
b	0
f	0
cc	3
nop	2

import json

from aiohttp import web

from .base import BaseJsonRpcServer
from .. import errors, middlewares, protocol, utils


__all__ = (
    'JsonRpcServer',
    'rpc_server',
)


class JsonRpcServer(BaseJsonRpcServer):
    async def handle_http_request(self, http_request: web.Request) -> web.Response:
        if http_request.method != 'POST':
            raise web.HTTPMethodNotAllowed(method=http_request.method, allowed_methods=('POST',))

        try:
            input_data = await http_request.json()
        except json.JSONDecodeError as e:
            response = protocol.JsonRpcResponse(error=errors.ParseError(utils.get_exc_message(e)))
            return web.json_response(response.dump(), dumps=self.json_serialize)

        output_data = await self._process_input_data(input_data, context={'http_request': http_request})

        return web.json_response(output_data, dumps=self.json_serialize)


rpc_server = JsonRpcServer(
    middlewares=middlewares.DEFAULT_MIDDLEWARES,
)


1			import json
2
3			from aiohttp import web
4
5			from .base import BaseJsonRpcServer
6			from .. import errors, middlewares, protocol, utils
7
8
9			__all__ = (
10			'JsonRpcServer',
11			'rpc_server',
12			)
13
14
15			class JsonRpcServer(BaseJsonRpcServer):
16			async def handle_http_request(self, http_request: web.Request) -> web.Response:
17			if http_request.method != 'POST':
18			raise web.HTTPMethodNotAllowed(method=http_request.method, allowed_methods=('POST',))
19
20			try:
21			input_data = await http_request.json()
22			except json.JSONDecodeError as e:
23			response = protocol.JsonRpcResponse(error=errors.ParseError(utils.get_exc_message(e)))
24			return web.json_response(response.dump(), dumps=self.json_serialize)
25
26			output_data = await self._process_input_data(input_data, context={'http_request': http_request})
27
28			return web.json_response(output_data, dumps=self.json_serialize)
29
30
31			rpc_server = JsonRpcServer(
32			middlewares=middlewares.DEFAULT_MIDDLEWARES,
33			)
34

expert-m / aiohttp-rpc

JsonRpcServer.handle_http_request() A last analyzed 2023-10-15 08:14 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

JsonRpcServer.handle_http_request() A
last analyzed 2023-10-15 08:14 UTC