aiohttp_rpc.client.http.JsonRpcClient.send_json() - Code Metrics - Inspection of "Split file client.py" - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 9a8b07...1de29a )

by Michael

created 2020-05-28 11:33 UTC

aiohttp_rpc.client.http.JsonRpcClient.send_json() A

↳ Parent: aiohttp_rpc.client.http

Complexity

Conditions

Size

Total Lines	14
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	14
rs	9.85
c	0
b	0
f	0
cc	3
nop	4

import typing

import aiohttp

from .base import BaseJsonRpcClient
from .. import errors, utils


__all__ = (
    'JsonRpcClient',
)


class JsonRpcClient(BaseJsonRpcClient):
    url: str
    session: typing.Optional[aiohttp.ClientSession]
    request_kwargs: dict
    _is_outer_session: bool

    def __init__(self,
                 url: str, *,
                 session: typing.Optional[aiohttp.ClientSession] = None,
                 **request_kwargs) -> None:
        self.url = url
        self.session = session
        self.request_kwargs = request_kwargs
        self._is_outer_session = session is not None

    async def connect(self) -> None:
        if not self.session:
            self.session = aiohttp.ClientSession(json_serialize=self.json_serialize)

    async def disconnect(self) -> None:
        if not self._is_outer_session:
            await self.session.close()

    async def send_json(self,
                        data: typing.Any, *,
                        without_response: bool = False) -> typing.Tuple[typing.Any, typing.Optional[dict]]:
        http_response = await self.session.post(self.url, json=data, **self.request_kwargs)

        try:
            json_response = await http_response.json()
        except aiohttp.ContentTypeError as e:
            raise errors.ParseError(utils.get_exc_message(e)) from e

        if without_response:
            return None, None

        return json_response, {'http_response': http_response}


1			import typing
2
3			import aiohttp
4
5			from .base import BaseJsonRpcClient
6			from .. import errors, utils
7
8
9			__all__ = (
10			'JsonRpcClient',
11			)
12
13
14			class JsonRpcClient(BaseJsonRpcClient):
15			url: str
16			session: typing.Optional[aiohttp.ClientSession]
17			request_kwargs: dict
18			_is_outer_session: bool
19
20			def __init__(self,
21			url: str, *,
22			session: typing.Optional[aiohttp.ClientSession] = None,
23			**request_kwargs) -> None:
24			self.url = url
25			self.session = session
26			self.request_kwargs = request_kwargs
27			self._is_outer_session = session is not None
28
29			async def connect(self) -> None:
30			if not self.session:
31			self.session = aiohttp.ClientSession(json_serialize=self.json_serialize)
32
33			async def disconnect(self) -> None:
34			if not self._is_outer_session:
35			await self.session.close()
36
37			async def send_json(self,
38			data: typing.Any, *,
39			without_response: bool = False) -> typing.Tuple[typing.Any, typing.Optional[dict]]:
40			http_response = await self.session.post(self.url, json=data, **self.request_kwargs)
41
42			try:
43			json_response = await http_response.json()
44			except aiohttp.ContentTypeError as e:
45			raise errors.ParseError(utils.get_exc_message(e)) from e
46
47			if without_response:
48			return None, None
49
50			return json_response, {'http_response': http_response}
51

expert-m / aiohttp-rpc

Push — master ( 9a8b07...1de29a )

aiohttp_rpc.client.http.JsonRpcClient.send_json() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like