Function Call

Using Throttled to check if a request is allowed is very simple.

You just need to call the Throttled.limit method and pass in the specified key, which will return a RateLimitResult object.

It is important to note that Throttled.limit does not raise any exceptions, you can determine whether the request is allowed by checking the RateLimitResult.limited attribute.

You can also get a snapshot of the Throttled state after calling Throttled.limit through the RateLimitResult.state attribute.

If you just want to check the latest state of Throttled without deducting requests, you can use the Throttled.peek method, which will also return a RateLimitState object.

The following example will guide you through the basic usage of Throttled:

Sync

from throttled import Throttled


def main() -> None:
    # By Default, it initializes a rate limiter with In-Memory,
    # allowing 60 requests per minute, using the token bucket algorithm.
    # Default: In-Memory storage, Token Bucket algorithm, 60 reqs / min.
    throttle = Throttled()

    # Consume 1 token.
    result = throttle.limit("key")
    # Should not be limited.
    assert not result.limited

    # Get the state of the rate limiter:
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0))
    print(result.state)

    # You can also get the state by using the `peek` method.
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0)
    print(throttle.peek("key"))

    # You can also specify the cost of the request.
    result = throttle.limit("key", cost=60)
    # This will consume 60 tokens, which exceeds the limit of 60 tokens per minute.
    assert result.limited

    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=1))
    print(result.state)


if __name__ == "__main__":
    main()

Async

import asyncio

from throttled.asyncio import Throttled


async def main() -> None:
    # By Default, it initializes a rate limiter with In-Memory,
    # allowing 60 requests per minute, using the token bucket algorithm.
    # Default: In-Memory storage, Token Bucket algorithm, 60 reqs / min.
    throttle = Throttled()

    # Consume 1 token.
    result = await throttle.limit("key")
    # Should not be limited.
    assert not result.limited

    # Get the state of the rate limiter:
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0))
    print(result.state)

    # You can also get the state by using the `peek` method.
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0)
    print(await throttle.peek("key"))

    # You can also specify the cost of the request.
    result = await throttle.limit("key", cost=60)
    # This will consume 60 tokens, which exceeds the limit of 60 tokens per minute.
    assert result.limited

    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=1))
    print(result.state)


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