Aiohttp GET request simplified.

aiohttp_helper.py

"""
Make aiohttp GET request like a champ
"""
# Author: Github.com/code-rgb

import asyncio
import aiohttp
from typing import Optional, Union, Dict

try:
    import ujson as json
except ModuleNotFoundError:
    import json


class AioHttp:
    def __init__(
        self, timeout: Optional[Union[int, float, aiohttp.ClientTimeout]] = 30
    ):
        self._session: Optional[aiohttp.ClientSession] = None
        self.timeout = timeout

    def get_new_session(self) -> aiohttp.ClientSession:
        return aiohttp.ClientSession(json_serialize=json.dumps)

    @property
    def session(self) -> Optional[aiohttp.ClientSession]:
        if self._session is None or self._session.closed:
            self._session = self.get_new_session()
        return self._session

    async def request(
        self,
        url: str,
        mode: str,
        params: Optional[Dict] = None,
        headers: Optional[Dict] = None,
        timeout: Optional[Union[int, float, aiohttp.ClientTimeout]] = None,
    ):
        """Perform HTTP GET request.

        Parameters:
            url (``str``):
                HTTP URL.

            mode (``str``):
                *"status"* - Returns response status,
                *"redirect"* - Returns Redirected URL,
                *"headers"* - Returns headers,
                *"json"* - Read and decodes JSON response (dict),
                *"text"* - Read response payload and decode (str),
                *"read"* - Read response payload (bytes)

            headers (``dict``, *optional*):
                Headers to be used while making the request.

            params (``dict``, *optional*):
                Optional parameters.

            timeout (``int`` | ``float`` | ``aiohttp.ClientTimeout``, *optional*):
                Timeout in seconds. This will override default timeout of 30 sec.

        Example:
            .. code-block:: python

            # initialize the class once
            http = AioHttp()
            ---
            import asyncio
            from . import http

            async def get_():
                r = await http.request("http://python.org", mode="text"):
                print(r)

            async def main():
                for i in range(7):
                    await get_()

                # Closing session before quitting
                await http.close()


            loop = asyncio.get_event_loop()
            loop.run_until_complete(main())
        """
        timeout_ = timeout or self.timeout
        try:
            async with self.session.get(
                url,
                timeout=timeout_,
                params=params,
                headers=headers,
            ) as resp:
                if mode == "status":
                    return resp.status
                if mode == "redirect":
                    return resp.url
                if mode == "headers":
                    return resp.headers
                if resp.status != 200:
                    print(
                        f"HTTP ERROR: '{url}' Failed with Status Code - {resp.status}"
                    )
                    return
                else:
                    if mode == "json":
                        try:
                            out = await resp.json()
                        except ValueError:
                            # catch json.decoder.JSONDecodeErrorr in case content type is not "application/json".
                            try:
                                out = json.loads(await resp.text())
                            except Exception as err:
                                print(
                                    f"{err}: Unable to get '{url}', {resp.headers['content-type']}"
                                )
                                return
                    elif mode == "text":
                        out = await resp.text()
                    elif mode == "read":
                        out = await resp.read()
                    else:
                        print(f"ERROR: Invalid Mode - '{mode}'")
                        return
        except asyncio.TimeoutError:
            print(f"Timeout ERROR: '{url}' Failed to Respond in Time ({timeout_} s).")
            return
        return out

    async def close(self):
        """
        Close aiohttp client session
        """
        await self.session.close()

😅😅