HTTP client

http-client.py

import aiohttp
import asyncio
import logging
import traceback
import uuid
from chardet import detect
from typing import Any, Optional, Tuple

logger = logging.getLogger(__name__)

RequestID = Optional[uuid.UUID]
ResponseBody = Optional[Any]


class HTTPResponse:

    def __init__(
        self,
        request_id: RequestID = None,  # HTTP request ID
        status: Optional[int] = None,  # HTTP status code
        body: ResponseBody = None,  # HTTP response body
        ok: bool = False,  # Is successful or not?
        exception: Exception = None
    ):
        self.request_id = request_id
        self.status = status
        self.body = body
        self.ok = ok
        self.exception = exception

    def get(self) -> Tuple[bool, ResponseBody]:
        return self.ok, self.body

    @property
    def is_empty(self) -> bool:
        return self.status == 204

    @property
    def is_timeout(self) -> bool:
        return isinstance(self.exception, asyncio.TimeoutError)

    def __str__(self) -> str:
        return (f'{self.__class__.__name__}('
                f'request_id={self.request_id}, '
                f'status={self.status}, '
                f'body={self.body}, '
                f'ok={self.ok}, '
                f'exception={self.exception})')

    def __repr__(self) -> str:
        return str(self)


async def request(
    method,
    url,
    output_format='json',
    log_level='info',
    propagate_exc=False,
    timeout=10,
    **kwargs
) -> HTTPResponse:
    session_params, request_params = {'timeout': aiohttp.ClientTimeout(total=timeout)}, {}

    request_id = uuid.uuid4()

    if log_level in ['debug']:
        logger.debug(f'request() -> {method}, {url}, {kwargs}')

    try:
        async with aiohttp.ClientSession(**session_params) as session:
            if method == 'get':
                handler = session.get
            elif method == 'post':
                handler = session.post
            else:
                handler = getattr(session, method)
            try:
                async with handler(url, **kwargs) as response:
                    try:
                        if output_format == 'raw':
                            body = await response.read()
                        elif output_format == 'text':
                            contents = await response.read()
                            encoding = detect(contents).get('encoding') or 'utf-8'
                            body = str(contents, encoding=encoding)
                        else:
                            body = await response.json()
                    except Exception as e:
                        traceback.print_exc()
                        logger.exception(e)

                        return HTTPResponse(request_id=request_id, status=response.status, ok=False, exception=e)

                    if log_level in ['debug']:
                        logger.debug(f'request() -> {method}, {url}, {kwargs}, {body}')

                    return HTTPResponse(request_id=request_id, status=response.status, ok=True, body=body)

            except Exception as e:
                traceback.print_exc()
                logger.exception(e)

                if propagate_exc:
                    raise e

                return HTTPResponse(request_id=request_id, status=response.status, ok=False, exception=e)

    except Exception as e:
        traceback.print_exc()
        logger.exception(e)

        if propagate_exc:
            raise e

        return HTTPResponse(request_id=request_id, status=response.status, ok=False, exception=e)