or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.mdretry-strategies.md
tile.json

tessl/pypi-aiohttp-retry

Simple retry client for aiohttp with configurable backoff strategies and error handling

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/aiohttp-retry@2.9.x

To install, run

npx @tessl/cli install tessl/pypi-aiohttp-retry@2.9.0

index.mddocs/

aiohttp-retry

A simple retry client for aiohttp that enables automatic retry functionality with various backoff strategies for HTTP requests. It offers configurable retry options including exponential backoff, random retry intervals, Fibonacci sequences, and custom timeout logic, with support for retrying on specific HTTP status codes, exceptions, or custom response evaluation callbacks.

Package Information

  • Package Name: aiohttp-retry
  • Package Type: pypi
  • Language: Python
  • Installation: pip install aiohttp-retry

Core Imports

from aiohttp_retry import RetryClient, ExponentialRetry

Common imports for different retry strategies:

from aiohttp_retry import (
    RetryClient,
    ExponentialRetry,
    RandomRetry,
    ListRetry,
    FibonacciRetry,
    JitterRetry,
    RequestParams
)

Type imports:

from aiohttp_retry.types import ClientType

Basic Usage

from aiohttp_retry import RetryClient, ExponentialRetry

async def main():
    # Basic usage with default exponential retry
    retry_client = RetryClient()
    async with retry_client.get('https://httpbin.org/status/503') as response:
        print(f"Status: {response.status}")
    
    await retry_client.close()

    # Using custom retry options
    retry_options = ExponentialRetry(
        attempts=5,
        start_timeout=0.5,
        max_timeout=10.0,
        factor=2.0
    )
    
    async with RetryClient(retry_options=retry_options) as client:
        async with client.get('https://httpbin.org/status/500') as response:
            print(f"Status: {response.status}")

    # Using with existing aiohttp ClientSession
    from aiohttp import ClientSession
    
    session = ClientSession()
    retry_client = RetryClient(client_session=session)
    
    response = await retry_client.get('https://httpbin.org/get')
    print(f"Status: {response.status}")
    
    await session.close()

Architecture

aiohttp-retry is built around two core concepts:

  • RetryClient: A wrapper around aiohttp's ClientSession that adds retry functionality
  • Retry Strategies: Pluggable strategies that determine timeout intervals between retry attempts

The library maintains full compatibility with aiohttp's async/await patterns and provides comprehensive tracing capabilities for monitoring retry attempts and debugging network communication issues.

Capabilities

HTTP Client Operations

The main RetryClient class provides all standard HTTP methods with automatic retry functionality. It supports all HTTP verbs and maintains compatibility with aiohttp's ClientSession interface.

class RetryClient:
    def __init__(
        self,
        client_session: ClientSession | None = None,
        logger: _LoggerType | None = None,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool = False,
        *args: Any,
        **kwargs: Any
    ): ...
    
    def get(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def post(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def put(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def patch(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def delete(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def head(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def options(
        self,
        url: _URL_TYPE,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def request(
        self,
        method: str,
        url: StrOrURL,
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any
    ) -> _RequestContext: ...
    
    def requests(
        self,
        params_list: list[RequestParams],
        retry_options: RetryOptionsBase | None = None,
        raise_for_status: bool | None = None
    ) -> _RequestContext: ...
    
    async def close(self) -> None: ...
    
    async def __aenter__(self) -> RetryClient: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...
    
    @property
    def retry_options(self) -> RetryOptionsBase: ...

Multi-Request Configuration

Advanced request configuration that allows different parameters for each retry attempt, enabling dynamic URL switching, header modification, and parameter adjustment between attempts.

from dataclasses import dataclass

@dataclass
class RequestParams:
    method: str
    url: _RAW_URL_TYPE
    headers: dict[str, Any] | None = None
    trace_request_ctx: dict[str, Any] | None = None
    kwargs: dict[str, Any] | None = None

Base Retry Strategy

Abstract base class that defines the interface for all retry strategies. Custom retry strategies can be implemented by inheriting from this class.

class RetryOptionsBase:
    def __init__(
        self,
        attempts: int = 3,
        statuses: Iterable[int] | None = None,
        exceptions: Iterable[type[Exception]] | None = None,
        methods: Iterable[str] | None = None,
        retry_all_server_errors: bool = True,
        evaluate_response_callback: Callable[[ClientResponse], Awaitable[bool]] | None = None
    ): ...
    
    def get_timeout(self, attempt: int, response: ClientResponse | None = None) -> float: ...

Retry Strategies

Type Definitions

from typing import Union, List, Tuple, Callable, Awaitable, Any, Iterable, Generator, Protocol
import logging
from aiohttp import ClientSession, ClientResponse
from aiohttp.typedefs import StrOrURL
from yarl import URL as YARL_URL

# Main client type alias
ClientType = Union[ClientSession, RetryClient]

# URL type definitions
_RAW_URL_TYPE = Union[StrOrURL, YARL_URL]
_URL_TYPE = Union[_RAW_URL_TYPE, List[_RAW_URL_TYPE], Tuple[_RAW_URL_TYPE, ...]]

# Logger protocol and type definition
class _Logger(Protocol):
    """Logger protocol defining required methods."""
    def debug(self, msg: str, *args: Any, **kwargs: Any) -> None: ...
    def warning(self, msg: str, *args: Any, **kwargs: Any) -> None: ...
    def exception(self, msg: str, *args: Any, **kwargs: Any) -> None: ...

_LoggerType = Union[_Logger, logging.Logger]

# Callback type definitions
EvaluateResponseCallbackType = Callable[[ClientResponse], Awaitable[bool]]
RequestFunc = Callable[..., Awaitable[ClientResponse]]

# Internal request context type (returned by HTTP methods)
class _RequestContext:
    """
    Internal request context that supports async context manager protocol.
    Returned by all HTTP methods on RetryClient.
    """
    def __await__(self) -> Generator[Any, None, ClientResponse]: ...
    async def __aenter__(self) -> ClientResponse: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...