-
Scout Monitoring
Free Django app performance insights with Scout Monitoring. Get Scout setup in minutes, and let us sweat the small stuff. A couple lines in settings.py is all you need to start monitoring your apps. Sign up for our free tier today.
ratelimit-headers has a test implementation of this draft.
class RateLimitTransport(httpx.AsyncHTTPTransport): def __init__(self, max_per_second: float = 5, **kwargs) -> None: """ Async HTTP transport with rate limit. Args: max_per_second: Maximum number of requests per second. Other args are passed to httpx.AsyncHTTPTransport. """ self.interval = 1 / max_per_second self.next_start_time = 0 super().__init__(**kwargs) async def notify_task_start(self): """ https://github.com/florimondmanca/aiometer/blob/358976e0b60bce29b9fe8c59807fafbad3e62cbc/src/aiometer/_impl/meters.py#L57 """ loop = asyncio.get_running_loop() while True: now = loop.time() next_start_time = max(self.next_start_time, now) until_now = next_start_time - now if until_now <= self.interval: break await asyncio.sleep(max(0, until_now - self.interval)) self.next_start_time = max(self.next_start_time, now) + self.interval async def handle_async_request(self, request: httpx.Request) -> httpx.Response: await self.notify_task_start() return await super().handle_async_request(request) async def __aenter__(self) -> Self: await self.notify_task_start() return await super().__aenter__() async def __aexit__(self, *args: Any) -> None: await super().__aexit__(*args)
There are already some implementations for Python HTTP clients. One of them is aiometer. But it's not suitable for my use case. Since httpx already has the internal pool, it would be better to reuse the design.