aiohttpx

Description

aiohttpx is an asynchronous HTTP client that merges the ergonomics and powerful API of httpx with the high-performance transport layer of aiohttp. It also provides optional Redis-powered caching and rate-limiting to enable efficient, production-grade request handling with minimal setup.

Features

• Fully asynchronous HTTP client using aiohttp as the transport.
• Optional Redis-based caching to reduce redundant API calls.
• Optional Redis-based rate limiting to control request throughput.
• Familiar API interface inspired by httpx.

Requirements

• Python 3.13 or higher
• Redis server (if using caching or rate limiting)

Installation

Using uv Tool

This project supports dependency management via the uv tool. To set up the project:

1. Install uv

   curl -LsSf https://astral.sh/uv/install.sh | sh
   

2. Add to the repository

   uv add https://git.meowly.ru/Miwory/aiohttpx.git
   

Configuration

aiohttpx supports several optional parameters for caching and rate limiting:

key — Redis prefix

A string used as the Redis key namespace for all cache and rate-limit entries. This allows multiple clients or services to share the same Redis instance without collisions.

limit — Rate limit

The maximum number of requests allowed per second for this client. This value is enforced using Redis, making it safe to use across distributed systems.

X-Cache-TTL — Enable caching for a request

To enable caching for a specific request, include the header:

X-Cache-TTL: <seconds>

Example:

response = await client.get(
    "/users",
    headers={"X-Cache-TTL": "60"},  # cache this endpoint for 60 seconds
)

If this header is present and Redis is configured, the response will be cached for the specified duration.

Usage

Basic Example

from aiohttpx.client import AioHTTPXClient

class TwitchAPIClient(AioHTTPXClient):
    def __init__(
        self,
        redis_url: str,
        client_id: str,
        client_secret: str,
        redirect_uri: str,
    ):
        self.base_uri = 'https://api.twitch.tv/helix'
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri

        super().__init__(
            base_url=self.base_uri,
            headers={'Client-Id': self.client_id},
            redis_url=redis_url,
            key='twitch',      # Redis prefix
            limit=10,          # 10 requests per second
            logger='Twitch API',
        )

    async def test_endpoint(self):
        ...

Linting and Pre-commit Checks

This project uses pre-commit and ruff for linting and formatting. Run the linting process with:

poe lint

License

This project is licensed under the MIT License. See LICENSE for details.