Quickstart¶

Installation¶

Rate Control requires Python 3.8 or later to run.

The library is hosted on PyPI, you can install it using pip:

pip install rate-control

Basic usage¶

Suppose you want to design a client-side rate controller for accessing some external API, to prevent your application from receiving responses with code 429.

Rate Control can help you track your requests, and handle rate limit errors before they occur.

Warning

Due to network fluctuations and other external factors, it is not guaranteed that the usage of these rate controllers will free you from all rate limit-related errors when requesting an API. You should still have a plan to handle such errors, though they will be much more rare.

Below is a simple example of how you could use a rate limiter, to ensure that no more than two requests are handled each minute.

Note

All available bucket algorithms are detailed in the dedicated page.

from asyncio import run
from rate_control import Duration, FixedWindowCounter, RateLimit, RateLimiter

async def main() -> None:
    bucket = FixedWindowCounter(capacity=2, duration=Duration.MINUTE)
    async with RateLimiter(bucket) as rate_limiter:
        for _ in range(3):
            try:
                async with rate_limiter.request():
                    print('Request executed')
            except RateLimit:
                print('Request rejected')

run(main())

from trio import run
from rate_control import Duration, FixedWindowCounter, RateLimit, RateLimiter

async def main() -> None:
    bucket = FixedWindowCounter(capacity=2, duration=Duration.MINUTE)
    async with RateLimiter(bucket) as rate_limiter:
        for _ in range(3):
            try:
                async with rate_limiter.request():
                    print('Request executed')
            except RateLimit:
                print('Request rejected')

run(main)

from anyio import run
from rate_control import Duration, FixedWindowCounter, RateLimit, RateLimiter

async def main() -> None:
    bucket = FixedWindowCounter(capacity=2, duration=Duration.MINUTE)
    async with RateLimiter(bucket) as rate_limiter:
        for _ in range(3):
            try:
                async with rate_limiter.request():
                    print('Request executed')
            except RateLimit:
                print('Request rejected')

run(main)

Output¶

Request executed
Request executed
Request rejected

Rate Control has much more functionalities to offer. Keep browsing this documentation to discover all the flexibility that this library can provide.