Python RESTClient Class with asyncio and aiohttp

Question

I've seen Using asyncio and aiohttp in classes but could not really figure out how to apply this to my own code.

The main idea of this code is to inherit from this RESTClient class whenever I build a new API Client. I'll probably improve on this code later to add functions to run multiple requests asynchronously.

While creating this bit of code, I ran into a few problems with the asyncio loop that, to be honest, I know very little about. This might not be the most pythonistic way, please provide your feedback.

I've posted this code to validate with the Python community if the way I'm handling the ClientSession in this example is the right way.

import asyncio
import aiohttp
import json
from dataclasses import dataclass


@dataclass
class Config:
    verify_ssl: bool = True
    tcp_connections: int = 5


class RestClient:
    def __init__(self, config: Config = None) -> None:
        if not config:
            self.config = Config()
        else:
            assert isinstance(config, Config)
            self.config = config

        try:
            loop = asyncio.new_event_loop()
            loop.run(self._create_session())
        except:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(self._create_session())

    def __del__(self):
        try:
            loop = asyncio.new_event_loop()
            loop.run_until_complete(self._close_session())
        except:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(self._close_session())

    async def _create_session(self):
        self.con = aiohttp.TCPConnector(ssl=self.config.verify_ssl)
        self.session = aiohttp.ClientSession(connector=self.con)

    async def _close_session(self):
        await self.session.close()
        await self.con.close()

    def request(
        self,
        method: str,
        url: str,
        query_param: dict = None,
        headers: dict = None,
        body: json = None,
    ):
        """Performs an Async HTTP request.

        Args:
            method (str): request method ('GET', 'POST', 'PUT', 'DELETE').
            url (str): request url.
            query_param (dict or None): url query parameters.
            header (dict or None): request headers.
            body (json or None): request body in case of method POST or PUT.
        """
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(
            self.async_request(
                method=method,
                url=url,
                query_param=query_param,
                headers=headers,
                body=body,
            )
        )

    async def async_request(
        self,
        method: str,
        url: str,
        query_param: dict = None,
        headers: dict = None,
        body: json = None,
    ):
        """Performs an Async HTTP request.

        Args:
            method (str): request method ('GET', 'POST', 'PUT', 'DELETE').
            url (str): request url.
            query_param (dict or None): url query parameters.
            header (dict or None): request headers.
            body (json or None): request body in case of method POST or PUT.
        """
        assert isinstance(method, str)
        assert isinstance(url, str)

        method = method.upper()
        assert method in ("GET",)  # TODO: Add ("POST", "PUT", "DELETE")

        headers = headers or {}

        try:
            if method == "GET":
                async with self.session.get(
                    url, params=query_param, headers=headers
                ) as response:
                    return await response.text()
        except:
            raise


if __name__ == "__main__":
    client = RestClient()
    print(client.request(method="GET", url="https://httpbin.org/get"))

Answer 1

While creating this bit of code, I ran into a few problems with the asyncio loop that, to be honest, I know very little about.

You are using asyncio very inappropriately. You need to pick one either RestClient is synchronous or asynchronous. Since you're using asynchronous libraries you have to make the class asynchronous. Therefore most methods have to be async methods, and you cannot call asyncio.run at all in the class.

Additionally your class should be an asynchronous context manager by changing _create_session to __aenter__ and _close_session to __aexit__.

This makes your code significantly simpler:

import asyncio
import aiohttp
import json
from dataclasses import dataclass


@dataclass
class Config:
    verify_ssl: bool = True
    tcp_connections: int = 5


class RestClient:
    def __init__(self, config: Config = None) -> None:
        self.config = Config() if config is none else config
        assert isinstance(self.config, Config)

    async def __enter__(self):
        self.con = aiohttp.TCPConnector(ssl=self.config.verify_ssl)
        self.session = aiohttp.ClientSession(connector=self.con)

    async def __exit__(self, *args):
        await self.session.close()
        await self.con.close()

    async def request(
        self,
        method: str,
        url: str,
        query_param: dict = None,
        headers: dict = None,
        body: json = None,
    ):
        """Performs an Async HTTP request.

        Args:
            method (str): request method ('GET', 'POST', 'PUT', ).
            url (str): request url.
            query_param (dict or None): url query parameters.
            header (dict or None): request headers.
            body (json or None): request body in case of method POST or PUT.
        """
        assert isinstance(method, str)
        assert isinstance(url, str)

        method = method.upper()
        assert method in ("GET",)  # TODO: Add ("POST", "PUT", "DELETE")

        headers = headers or {}

        try:
            if method == "GET":
                async with self.session.get(
                    url, params=query_param, headers=headers
                ) as response:
                    return await response.text()
        except:
            raise


async def main():
    async with RestClient() as client:
        print(await client.request(method="GET", url="https://httpbin.org/get"))


if __name__ == "__main__":
    asyncio.run(main())

You can make a synchronous -> asynchronous class to build on-top of RestClient but it's generally not a good idea. If you don't want to have all your code be asynchronous then you shouldn't use async.

Answer 2

I've updated the code according to the feedback from @Peilonrayz (thanks!)

rest.py

import asyncio
import aiohttp
import json

from dataclasses import dataclass


@dataclass
class Config:
    verify_ssl: bool = True
    tcp_connections: int = 5


class RestClient:
    """A generic HTTP Rest client."""

    def __init__(self, config: Config = None) -> None:
        self.config = Config() if config is None else config
        assert isinstance(self.config, Config)

    async def __aenter__(self):
        self._con = aiohttp.TCPConnector(
            verify_ssl=self.config.verify_ssl, limit=self.config.tcp_connections
        )
        self._session = aiohttp.ClientSession(connector=self._con)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self._session.close()
        await self._con.close()

    async def request(
        self,
        method: str,
        url: str,
        query_param: dict = None,
        headers: dict = None,
        body: json = None,
    ):
        """Performs an Async HTTP request.

        Args:
            method (str): request method ('GET', 'POST', 'PUT', ).
            url (str): request url.
            query_param (dict or None): url query parameters.
            header (dict or None): request headers.
            body (json or None): request body in case of method POST or PUT.
        """
        assert isinstance(method, str)
        assert isinstance(url, str)

        method = method.upper()
        assert method in ("GET",)  # TODO: Add ("POST", "PUT", "DELETE")

        headers = headers or {}

        try:
            if method == "GET":
                async with self._session.get(
                    url, params=query_param, headers=headers
                ) as response:
                    return await response.text()
        except:
            raise
            # TODO: Handle exceptions explicitly


async def main():
    async with RestClient() as client:
        print(await client.request(method="GET", url="https://httpbin.org/get"))


if __name__ == "__main__":
    loop = asyncio.get_event_loop() #asyncio.run(main()) throws an error on windows.
    loop.run_until_complete(main())
```