Is this simple client-side rate-limiting middleware safe for small projects?

[loic-bellinger]

I’ve put together a small client-side rate-limiting middleware for aiohttp to throttle requests per host to a target RPS, with optional jitter. Before I rely on it, I’d love your feedback: is this approach sound for a small project, or are there hidden pitfalls that could break things with aiohttp? Also, is there a more official/recommended way to do this ?

Code (simplified)

import asyncio
import random
import time
from collections import defaultdict

from aiohttp import ClientHandlerType, ClientRequest, ClientResponse


class RateLimitMiddleware:
    """
    Very simple per-host RPS limiter with jitter.
    """

    def __init__(
        self,
        rate_limit_rules: dict[str, float],
        default_rps: float,
        jitter_factor: float,
    ):
        self._rate_limit_rules = rate_limit_rules
        self._default_rps = default_rps
        self._jitter_factor = jitter_factor
        self._host_to_rps_cache: dict[str, float] = {}

        self._locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
        self._next_allowed_time: defaultdict[str, float] = defaultdict(float)

    def _resolve_rps_for_host(self, host: str) -> float:
        # Simplified: real code maps host → rps using rules, fallback to default
        return self._rate_limit_rules.get(host, self._default_rps)

    async def __call__(
        self,
        request: ClientRequest,
        handler: ClientHandlerType,
    ) -> ClientResponse:
        host = request.url.host
        if not host:
            return await handler(request)

        rps = self._resolve_rps_for_host(host)
        if rps <= 0:
            return await handler(request)

        base_interval = 1.0 / rps

        if self._jitter_factor > 0:
            interval = base_interval * (1.0 + self._jitter_factor * random.random())
        else:
            interval = base_interval

        lock = self._locks[host]
        async with lock:
            now = time.monotonic()
            t = self._next_allowed_time[host]

            slot = max(now, t)
            self._next_allowed_time[host] = slot + interval

        sleep_duration = slot - now
        if sleep_duration > 0:
            await asyncio.sleep(sleep_duration)

        return await handler(request)

[Dreamsorcerer]

I think that's probably fine, except that the jitter doesn't guarantee the rate limit. e.g. If you do 5 rps, and the random.random() calls average <0.5 (i.e. 50% of the time), then you'll end up doing 6 rps and trigger the rate limit.

It also assumes you don't want to handle bursty behaviour (e.g. we've had cases where we have something like 10 requests per minute rate limit, but we get bursts of 5 requests we want to make. With your code, the 5th request would be delayed 30 seconds when it could have been made immediately).

[loic-bellinger]

You're right. Thanks a lot for your answer : I'll improve the code.

On a side note : would that make sense to add something similar to the /example repo ?

[Dreamsorcerer]

We have a cookbook, so adding to that might be a good contribution:
https://docs.aiohttp.org/en/stable/client_middleware_cookbook.html