aiohttp_rpc.client.http

Complexity

Total Complexity

8

Size/Duplication

Total Lines	51
Duplicated Lines	0 %

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	JsonRpcClient.send_json()	0	14	3
A	JsonRpcClient.__init__()	0	8	1
A	JsonRpcClient.disconnect()	0	3	2
A	JsonRpcClient.connect()	0	3	2

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}