httpx-limiter 0.5.0

A lightweight package that provides rate-limited httpx transports.

HTTPX Limiter

Package
Meta
Automation

A lightweight package that provides rate-limited httpx transports.

Installation

The package is published on PyPI. Install it, for example, with the aiolimiter backend.

pip install 'httpx-limiter[aiolimiter]'

There is also a pyrate-limiter backend available.

pip install 'httpx-limiter[pyrate]'

Rate Limiter Backends

This package provides two different asynchronous rate limiter implementations to choose from:

1. aiolimiter
2. pyrate-limiter

[!IMPORTANT] While both implementations fulfill similar purposes, there are significant differences between them. Please read the descriptions below to make an informed decision about which one best suits your needs.

1. aiolimiter

• Single rate limit only: Supports only one rate limit at a time
• Lightweight: Minimal dependencies and simpler configuration
• Linear token refresh rate: As an example, if you set a rate of 2 per second, roughly one token will be added every 500 milliseconds

from httpx_limiter.aiolimiter import AiolimiterAsyncLimiter

limiter = AiolimiterAsyncLimiter.create(Rate.create(magnitude=20))

2. pyrate-limiter

• Multiple rate limits: Support for multiple concurrent rate limits, for example, 10 requests per second and 100 requests per minute
• Flexible configuration: Comprehensive configuration options
• Multiprocessing dependency: Current implementation depends on multiprocessing which may not be available in all environments, such as pyodide
• Stepwise token refresh rate: As an example, if you set a rate of 2 per second, two token will be added every second

from httpx_limiter.pyrate import PyrateAsyncLimiter

# Single rate limit
limiter = PyrateAsyncLimiter.create(Rate.create(magnitude=20))

# Multiple rate limits
limiter = PyrateAsyncLimiter.create(
    Rate.create(magnitude=10),  # 10 per second
    Rate.create(magnitude=100, duration=60),  # 100 per minute
)

Tutorial

You can limit the number of requests made by an HTTPX client using the transports provided in this package. That is useful in situations when you need to make a large number of asynchronous requests against endpoints that implement a rate limit.

Single Rate Limit

The simplest use case is to apply a single rate limit to all requests. If you want to be able to make twenty requests per second, for example, use the following code:

import httpx
from httpx_limiter import AsyncRateLimitedTransport, Rate
from httpx_limiter.pyrate import PyrateAsyncLimiter

async def main():
    limiter = PyrateAsyncLimiter.create(Rate.create(magnitude=20))

    async with httpx.AsyncClient(
        transport=AsyncRateLimitedTransport.create(limiter=limiter),
    ) as client:
        response = await client.get("https://httpbin.org")

[!IMPORTANT] Due to limitations in the design of the underlying leaky bucket implementation, which is used to implement the rate limiting, the magnitude of the rate is also the maximum capacity of the bucket. That means, if you set a rate that is larger than one, a burst of requests equal to that capacity will be allowed. If you do not want to allow any bursts, set the magnitude to one, but the duration to the inverse of your desired rate. If you want to allow twenty requests per second, for example, set the magnitude to 1 and the duration to 0.05 seconds.

Rate.create(magnitude=1, duration=1/20)

Multiple Rate Limits

For more advanced use cases, you can apply different rate limits based on a concrete implementation of the AbstractRateLimiterRepository. There are two relevant methods that both get passed the current request. One method needs to return an identifier for which rate limiter to choose, and the other method creates a rate limiter. See the following example:

import httpx
from httpx_limiter import (
    AbstractRateLimiterRepository,
    AsyncMultiRateLimitedTransport,
    Rate
)
from httpx_limiter.aiolimiter import AiolimiterAsyncLimiter

class DomainBasedRateLimiterRepository(AbstractRateLimiterRepository):
    """Apply different rate limits based on the domain being requested."""

    def get_identifier(self, request: httpx.Request) -> str:
        """Return the domain as the identifier for rate limiting."""
        return request.url.host

    def create(self, request: httpx.Request) -> AiolimiterAsyncLimiter:
        """Create a rate limiter for the domain."""
        return AiolimiterAsyncLimiter.create(Rate.create(magnitude=25))

client = httpx.AsyncClient(
    transport=AsyncMultiRateLimitedTransport.create(
        repository=DomainBasedRateLimiterRepository(),
    ),
)

[!TIP] You are free to ignore the request parameter and use global information like the time of day to determine the rate limit.

from datetime import datetime, timezone

import httpx
from httpx_limiter import AbstractRateLimiterRepository, Rate
from httpx_limiter.pyrate import PyrateAsyncLimiter

class DayNightRateLimiterRepository(AbstractRateLimiterRepository):
    """Apply different rate limits based on the time of day."""

    def get_identifier(self, request: httpx.Request) -> str:
        """Identify whether it is currently day or night."""
        if 6 <= datetime.now(tz=timezone.utc).hour < 18:
            return "day"

        return "night"

    def create(self, request: httpx.Request) -> PyrateAsyncLimiter:
        """Create a rate limiter based on the time of day."""
        if self.get_identifier(request) == "day":
            return PyrateAsyncLimiter.create(Rate.create(magnitude=10))

        return PyrateAsyncLimiter.create(Rate.create(magnitude=100))

Copyright

• Copyright © 2024, 2025 Moritz E. Beber.
• Free software distributed under the Apache Software License 2.0.