Python aiohttp extension

Question

I made my first open-source project and wonder if I could do it better. I'll give briefly description, ask some questions and then post whole code. Open to any opinons and suggestions, thank you.

So, Aiohttp is asynchronous HTTP Client for Python. You can find it's documentation here. However all you need to know is that it have ClientSession class for making responses. This class is made in async-with approach.
Developers suggest to use it like this:

async def fetch(session, url):
  async with session.get(url) as response:
    return await response.text()

async with aiohttp.ClientSession() as session:
  html = await fetch(session, 'http://python.org')
  print(html)

I made an extension to aiohttp for retries. It's common idea to retry a request in case it failed. I tried to save this async-with approach. So, here two classes:

• RetryClient - client for retries
  It takes the same params as aiohttp ClientSession and do the same things. The only difference: methods like 'get' takes additional params like number of attemps.

• _RequestContext - async-with context for handling request
  Actually it contains all retry logic and tries to do requests. When it do request, it send it to aiohttp and then handle response in proper way.

In my opinion the main problem here is huge usage of **kwargs. All params are stored here. I tested that it will be broken on wrong param, but smart IDE like Pycharm wouldn't suggest you anything. However, I don't know how to fix it.

You can find the whole project here:
https://github.com/inyutin/aiohttp_retry

The whole code:

import asyncio
import logging
from aiohttp import ClientSession, ClientResponse
from typing import Any, Callable, Optional, Set, Type

# Options
_RETRY_ATTEMPTS = 3
_RETRY_START_TIMEOUT = 0.1
_RETRY_MAX_TIMEOUT = 30
_RETRY_FACTOR = 2


class _RequestContext:
    def __init__(self, request: Callable[..., Any],  # Request operation, like POST or GET
                 url: str,  # Just url
                 retry_attempts: int = _RETRY_ATTEMPTS,  # How many times we should retry
                 retry_start_timeout: float = _RETRY_START_TIMEOUT,  # Base timeout time, then it exponentially grow
                 retry_max_timeout: float = _RETRY_MAX_TIMEOUT,  # Max possible timeout between tries
                 retry_factor: float = _RETRY_FACTOR,  # How much we increase timeout each time
                 retry_for_statuses: Optional[Set[int]] = None,  # On which statuses we should retry
                 retry_exceptions: Optional[Set[Type]] = None,  # On which exceptions we should retry
                 **kwargs: Any
                 ) -> None:
        self._request = request
        self._url = url

        self._retry_attempts = retry_attempts
        self._retry_start_timeout = retry_start_timeout
        self._retry_max_timeout = retry_max_timeout
        self._retry_factor = retry_factor

        if retry_for_statuses is None:
            retry_for_statuses = set()
        self._retry_for_statuses = retry_for_statuses

        if retry_exceptions is None:
            retry_exceptions = set()
        self._retry_exceptions = retry_exceptions

        self._kwargs = kwargs

        self._current_attempt = 0
        self._response: Optional[ClientResponse] = None

    def _exponential_timeout(self) -> float:
        timeout = self._retry_start_timeout * (self._retry_factor ** (self._current_attempt - 1))
        return min(timeout, self._retry_max_timeout)

    def _check_code(self, code: int) -> bool:
        return 500 <= code <= 599 or code in self._retry_for_statuses

    async def _do_request(self) -> ClientResponse:
        try:
            self._current_attempt += 1
            response: ClientResponse = await self._request(self._url, **self._kwargs)
            code = response.status
            if self._current_attempt < self._retry_attempts and self._check_code(code):
                retry_wait = self._exponential_timeout()
                await asyncio.sleep(retry_wait)
                return await self._do_request()
            self._response = response
            return response

        except Exception as e:
            retry_wait = self._exponential_timeout()
            if self._current_attempt < self._retry_attempts:
                for exc in self._retry_exceptions:
                    if isinstance(e, exc):
                        await asyncio.sleep(retry_wait)
                        return await self._do_request()

            raise e

    async def __aenter__(self) -> ClientResponse:
        return await self._do_request()

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        if self._response is not None:
            if not self._response.closed:
                self._response.close()


class RetryClient:
    def __init__(self, logger: Any = None, *args: Any, **kwargs: Any) -> None:
        self._client = ClientSession(*args, **kwargs)
        self._closed = False

        if logger is None:
            logger = logging.getLogger("aiohttp_retry")

        self._logger = logger

    def __del__(self) -> None:
        if not self._closed:
            self._logger.warning("Aiohttp retry client was not closed")

    @staticmethod
    def _request(request: Callable[..., Any], url: str, **kwargs: Any) -> _RequestContext:
        return _RequestContext(request, url, **kwargs)

    def get(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.get, url, **kwargs)

    def options(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.options, url, **kwargs)

    def head(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.head, url, **kwargs)

    def post(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.post, url, **kwargs)

    def put(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.put, url, **kwargs)

    def patch(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.patch, url, **kwargs)

    def delete(self, url: str, **kwargs: Any) -> _RequestContext:
        return self._request(self._client.delete, url, **kwargs)

    async def close(self) -> None:
        await self._client.close()
        self._closed = True

    async def __aenter__(self) -> 'RetryClient':
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

Answer 1

• I think RetryClient is a bad name as it doesn't highlight that you're wrapping a session.

• Taking lots of retry_* parameters screams to me that you should make a class.

  For example popular libraries like requests and urllib3 use a Retry class.

• You can pass a tuple to isinstance to check against multiple types in one call.

• You use a tuple with except to filter to only those exceptions. Removing the need for isinstance at all.
• You can make _exponential_timeout a function that makes a full-fledged itarable.

• Making the Retry class take an iterable on how long each delay should be allows for easy customization from users.

  Want a 3 gap 20 times is easy:

  Retry(timeouts=[3] * 20)
  

  Users can also make it so it retries infinitely.

  class InfinateOnes:
      def __iter__(self):
          while True:
              yield 1
  

• I would prefer to see an explicit while True and iteration over recursion.

• I would prefer if Retry was designed in a way in which it would work with any function that returns a ClientResponse.

  A part of me that likes to make things as generic as possible, would prefer changing statuses to a callback to check if the return is valid. However this doesn't make too much sense in a bespoke library.

In all I think this drastically simplifies _RequestContext.
^{Note: untested}

import asyncio
import logging
from aiohttp import ClientSession, ClientResponse
from typing import Any, Callable, Optional, Set, Type, Iterable

# Options
_RETRY_ATTEMPTS = 3
_RETRY_START_TIMEOUT = 0.1
_RETRY_MAX_TIMEOUT = 30
_RETRY_FACTOR = 2


def exponential(
    attempts: int = _RETRY_ATTEMPTS,
    start: int = _RETRY_START_TIMEOUT,
    maximum: int = _RETRY_MAX_TIMEOUT,
    factor: int = _RETRY_FACTOR,
) -> Iterable[float]:
    return [
        min(maximum, start * (factor ** i))
        for i in range(attempts)
    ]


class Retry:
    def __init__(
        self,
        timeouts: Iterable[float] = exponential(),
        statuses: Optional[Set[int]] = None,
        exceptions: Optional[Tuple[Type]] = None,
    ) -> None:
        self._timeouts = timeouts
        self._statuses = statuses or set()
        self._exceptions = exceptions or ()

    def _is_retry_status(self, code):
        return 500 <= code <= 599 or code in self._statuses

    async def retry(
        self,
        callback: Callable[[...], ClientResponse],
        *args,
        **kwargs,
    ) -> ClientResponse:
        timeouts = iter(self.timeouts)
        while True:
            try:
                response = await self._request(*args, **kwargs)
                if not self._is_retry_status(response.status):
                    return response
                try:
                    retry_wait = next(timouts)
                except StopIteration:
                    return response
            except self._retry_exceptions as e:
                try:
                    retry_wait = next(timouts)
                except StopIteration:
                    raise e from None
            await asyncio.sleep(retry_wait)


class _RequestContext:
    def __init__(
        self,
        request: Callable[..., Any],
        url: str,
        retry: Retry,
        **kwargs: Any
    ) -> None:
        self._request = request
        self._url = url
        self._kwargs = kwargs
        self._retry = retry

    async def __aenter__(self) -> ClientResponse:
        return await self._retry.retry(self._request, self._url, **self._kwargs)

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        if self._response is not None:
            if not self._response.closed:
                self._response.close()