Context Manager

You can use the context manager to limit the code block. When access is allowed, return RateLimitResult.

If the limit is exceeded or the retry timeout is exceeded, it will raise LimitedError.

Sync

from throttled import Throttled, exceptions


def call_api() -> None:
    print("doing something...")


def main() -> None:
    throttle: Throttled = Throttled(key="/api/v1/users/", quota="1/m")
    with throttle as result:
        # The first call will not be rate limited.
        assert not result.limited
        # Get the state of the rate limiter:
        # >> RateLimitState(limit=1, remaining=0, reset_after=60, retry_after=0)
        print(result.state)

        call_api()

    try:
        with throttle:
            call_api()
    except exceptions.LimitedError as exc:
        # >> Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
        print(exc)


if __name__ == "__main__":
    main()

Async

import asyncio

from throttled.asyncio import Throttled, exceptions


async def call_api() -> None:
    print("doing something...")


async def main() -> None:
    throttle: Throttled = Throttled(key="/api/v1/users/", quota="1/m")
    async with throttle as result:
        # The first call will not be rate limited.
        assert not result.limited
        # Get the state of the rate limiter:
        # >> RateLimitState(limit=1, remaining=0, reset_after=60, retry_after=0)
        print(result.state)

        await call_api()

    try:
        async with throttle:
            await call_api()
    except exceptions.LimitedError as exc:
        # >> Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
        print(exc)


if __name__ == "__main__":
    asyncio.run(main())