python rate limiting

rate_limiter.py

class RateLimiter:
    def __init__(
        self, *, name: str, max_requests_per_period: int, period_duration_secs: int
    ):
        self.__name = name
        self.__max_requests_per_period = max_requests_per_period
        self.__semaphore = asyncio.Semaphore(max_requests_per_period)
        self.__period_duration_secs = period_duration_secs

    T = TypeVar("T")

    async def call(self, coroutine: T | Awaitable[T]) -> T:
        print(
            f"Acquiring rate-limit token for {self.__name} (max {self.__max_requests_per_period} per {self.__period_duration_secs} secs)"
        )
        await self.__semaphore.acquire()
        asyncio.create_task(self.__wait_and_release())
        if isinstance(coroutine, Awaitable):
            return await coroutine
        return coroutine

    async def __wait_and_release(self):
        await asyncio.sleep(self.__period_duration_secs)
        self.__semaphore.release()


# Client-side rate limiter, since we currently have a 5x/minute rate limit on AWS Bedrock
# https://gloo-global.slack.com/archives/C03KV1PJ6EM/p1731353768431689
max_five_times_per_minute = RateLimiter(
    name="AWS Bedrock", max_requests_per_period=5, period_duration_secs=1
)