Rate Limiting with Python and Redis

To make sure that our services are running optimally, we prevent abuse by rate limiting. For example, if we have an API server, we limit the number of API calls per time period. One approach is to use Redis because we can easily scale our services. We often create multiple instances of our server and we put them behind load balancers. The Redis server creates a "glue" between our instances and makes sure that counters are "atomic" and that our servers are synced.

Redis in Python

To connect to a Redis server we need the module redis-py. Optionally, we will use hiredis-py as Redis reply parser. This is a wrapper to the hiredis C parser and more performant that the default redis-py parser. If installed, redis-py will use the hiredis parser.

pip install redis hiredis

Writing a Redis client

The following code shows how to create a Redis client. We initialize by calling the Redis() constructor and passing the parameters host, port, db, and password. To verify connection, we call the ping() method and should return True fo successful connection.

import os

import redis

redis_client = None


def get_redis_client():
    global redis_client
    if not redis_client:
        # get credentials from environment variables
        redis_client = redis.Redis(
            host=os.getenv('REDIS_HOST'),
            port=os.getenv('REDIS_PORT'),
            db=os.getenv('REDIS_DB'),
            password=os.getenv('REDIS_PASSWORD')
        )
    assert redis_client.ping()  # check if connection is successful
    return redis_client

The Naive Approach

For instance, we want to call the following function limited to 100 calls per second.

def my_function():
    pass  # do something

We can simply use the Redis command INCR. This will act as a counter. The name of the key should be unique per 1 second so the key name should be rate-limit:<timestamp>. To automatically remove the key from Redis we set the timeout of the key for 1 second by checking using TTL and by setting the timeout using EXPIRE.

def rate_per_second(function, count):
    client = get_redis_client()
    key = f"rate-limit:{int(time.time())}"
    if int(client.incr(key)) > count:
        raise RateLimitExceeded
    if client.ttl(key) == -1:  # timeout is not set
        client.expire(key, 1)  # expire in 1 second
    return function()

We created a custom exception RateLimitExceeded to be raised when the count reaches N.

class RateLimitExceeded(Exception):
    pass

The following code shows how to use the rate_per_second() function.

success = fail = 0
for i in range(2000):
    try:
        rate_per_second(my_function, 100)  # example: 100 requests per second
        success += 1
    except RateLimitExceeded:
        fail += 1
    time.sleep(5/1000)  # sleep every 5 milliseconds
print(f"Success count = {success}")
print(f"Fail count = {fail}")

Converting to Decorator

Rate limiting will be easier when using a decorator.

@rate_per_second(100)  # example: 100 requests per second
def my_function():
    pass  # do something

This works by converting the original rate_per_second() function to a decorator.

def rate_per_second(count):
    def _rate_per_second(function):
        def __rate_per_second(*args, **kwargs):
            client = get_redis_client()
            key = f"rate-limit:{int(time.time())}"
            if int(client.incr(key)) > count:
                raise RateLimitExceeded
            if client.ttl(key) == -1:  # timeout is not set
                client.expire(key, 1)  # expire in 1 second
            return function(*args, *kwargs)
        return __rate_per_second
    return _rate_per_second

Now we can simply call the function itself.

success = fail = 0
for i in range(2000):
    try:
        my_function()
        success += 1
    except RateLimitExceeded:
        fail += 1
    time.sleep(5/1000)  # sleep every 5 milliseconds
print(f"Success count = {success}")
print(f"Fail count = {fail}")

To Wrap Up

The approaches described above can be easily added to any Python application, for example, rate limiting calls to an API. One advantage of using Redis is its speed and making our services easily scalable.

The examples can be easily downloaded from Github.

Read more articles like this in the future by buying me a coffee!

Buy me a coffee