X-Rate-Limit headers

[Tomas2D]

Is there any way to automatically set rate limiting based on the following HTTP Headers:

For org-wide rate limits, the three headers show the limit that is being enforced, when it resets, and how close you are to hitting the limit:
X-Rate-Limit-Limit - the rate limit ceiling that is applicable for the current request.
X-Rate-Limit-Remaining - the number of requests left for the current rate-limit window.
X-Rate-Limit-Reset - the time at which the rate limit resets, specified in UTC epoch time (in seconds).
Source: https://developer.okta.com/docs/reference/rl-best-practices/

[lovelydinosaur]

Not automatically, but using a custom transport is a good approach for this kind of use-case...

class RateLimitTransport(httpx.HTTPTransport):
    def handle_request(self, request):
        # apply per-domain rate limits
        response = self.handle_request(request)
        # inspect response headers, update per-domain rate limits
        return response

client = httpx.Client(transport=RateLimitTransport())

https://www.python-httpx.org/advanced/#custom-transports

[Tomas2D]

Thanks for pointing out 👍🏻 . This is my working implementation ⬇️

import httpx
from aiolimiter import AsyncLimiter
from httpx import HTTPStatusError, Request, RequestError, Response

__all__ = ["AsyncRateLimiter", "AsyncRateLimitTransport", "AsyncRetryTransport"]

class AsyncRateLimiter(AsyncLimiter):
    def update_limit(self, *, max_rate: Optional[float] = None, time_period: Optional[float] = None):
        self.max_rate = max_rate or self.max_rate
        self.time_period = time_period or self.time_period
        self._rate_per_sec = self.max_rate / self.time_period

class AsyncRateLimitTransport(AsyncRetryTransport):
    def __init__(self, *args, default_max_rate: float, default_time_period: float, **kwargs):
        super().__init__(*args, **kwargs)
        self.limiter = AsyncRateLimiter(max_rate=default_max_rate, time_period=default_time_period)

    def update_rate_limit(self, response: Response):
        max_capacity = response.headers.get("x-ratelimit-limit")
        reset_period_in_seconds = response.headers.get("x-ratelimit-reset")

        if max_capacity and reset_period_in_seconds:
            self.limiter.update_limit(
                max_rate=max(1, int(max_capacity)),
                time_period=max(1, int(reset_period_in_seconds)),
            )

    async def handle_async_request(
        self,
        request: Request,
    ) -> Response:
        async with self.limiter:
            response = await super().handle_async_request(request)
            self.update_rate_limit(response)
            return response