Rate Limit a Python API with Redis®

Published on September 1, 2022•Updated on June 21, 2024

Introduction

Rate-limiting is a technology for controlling the number of requests a user can make to an Application Programming Interface (API) service in a given time frame. A good rate-limiter model enforces a fair-usage policy and prevents end-users from abusing business applications. If an application receives many requests without any control mechanism, the application can slow down or shut down completely.

This guide shows you how to implement a rate-limiter for a Python API with Redis® on Ubuntu 20.04.

Prerequisites

To follow along with this guide:

Deploy an Ubuntu 20.04 server.
Create a non-root sudo user.
SSH to your server as a non-root sudo user and install the following packages:
- Redis® server.
- MySQL server.

1. Set Up a Sample Database and User Account

This guide shows you how to code a sample Python API to help you better understand how the rate-limiting model works. This API returns a list of continents from a MySQL server in JSON format. The API only allows 5 requests per minute (60 seconds). The API uses Redis® to limit any users trying to exceed their quota. Follow the steps below to initialize the database and set up a MySQL user account:

Enter your password and press Enter to proceed. Then, run the SQL commands below to create a sample countries database and an api_user account. Replace EXAMPLE_PASSWORD with a secure value. Later, your Python API requires these details to access the database.

 mysql> CREATE DATABASE countries;
        CREATE USER 'api_user'@'localhost' IDENTIFIED WITH mysql_native_password BY 'EXAMPLE_PASSWORD';
        GRANT ALL PRIVILEGES ON countries.* TO 'api_user'@'localhost';
        FLUSH PRIVILEGES;

Output.

 ...
 Query OK, 0 rows affected (0.01 sec)

Switch to the new countries database.
```
 mysql> USE countries;
```
Output.
```
 Database changed
```
Create a continents table by running the following SQL command. This table stores a list of all continents in the world. Issue the AUTO_INCREMENT keyword to instruct MySQL to assign a continent_id for each record.
```
 mysql> CREATE TABLE continents (
            continent_id BIGINT NOT NULL PRIMARY KEY AUTO_INCREMENT,
            continent_name VARCHAR(100)
        ) ENGINE = InnoDB;
```
Output.
```
 Query OK, 0 rows affected (0.04 sec)
```

Populate the continents table with the seven continents.

 mysql> INSERT INTO continents (continent_name) VALUES ('ASIA');
        INSERT INTO continents (continent_name) VALUES ('AFRICA');
        INSERT INTO continents (continent_name) VALUES ('NORTH AMERICA');
        INSERT INTO continents (continent_name) VALUES ('SOUTH AMERICA');
        INSERT INTO continents (continent_name) VALUES ('ANTARCTICA');
        INSERT INTO continents (continent_name) VALUES ('EUROPE');
        INSERT INTO continents (continent_name) VALUES ('AUSTRALIA');

Output.

 ...
 Query OK, 1 row affected (0.01 sec)

Query the continents table to ensure the data is in place.

 mysql> SELECT
            continent_id,
            continent_name
        FROM continents;

Output.

 +--------------+----------------+
 | continent_id | continent_name |
 +--------------+----------------+
 |            1 | ASIA           |
 |            2 | AFRICA         |
 |            3 | NORTH AMERICA  |
 |            4 | SOUTH AMERICA  |
 |            5 | ANTARCTICA     |
 |            6 | EUROPE         |
 |            7 | AUSTRALIA      |
 +--------------+----------------+
 7 rows in set (0.00 sec)

Log out from the MySQL server.
```
 mysql> QUIT;
```
Output.
```
 Bye
```
Your countries database, api_user account, and the continents table are now ready.
Proceed to the next step to create an API that displays the continents in JSON format via the HTTP protocol using a web server.

2. Create a `Continents` Class

Coding your Python application in modules is a good practice that allows you to organize your code into small manageable pieces. For this application, you require a separate Continents class that connects to the MySQL server and queries the continents table before returning results in JSON format.

Before coding, it's conventional to separate your application's source code from system files. Follow the steps below to make a project directory and create a Continents class:

Create a new project directory.
```
 $ mkdir project
```
Navigate to the project directory.
```
 $ cd project
```
Use a text editor to open a new continents.py file.
```
 $ nano continents.py
```

Enter the following information into the continents.py file. Replace EXAMPLE_PASSWORD with the correct database password for the api_user.

 import mysql.connector
 import json

 class Continents:

     def get_db_connect(self):

         dbHost = "localhost"
         dbUser = "api_user"
         dbPass = "EXAMPLE_PASSWORD"
         dbName = "countries"

         db_con = mysql.connector.connect(host = dbHost, user = dbUser, password = dbPass, database = dbName)

         return db_con

     def get_continents(self):

         queryString = "select * from continents"

         dbConn = self.get_db_connect()

         dbCursor = dbConn.cursor(dictionary = True)

         dbCursor.execute(queryString)

         continentsList = dbCursor.fetchall()

         return json.dumps(continentsList , indent = 2)

Save and close the continents.py file.

The continents.py file explained:

The import mysql.connector declaration imports a Python driver for communicating with MySQL.
The import json statement imports a module for formatting strings to JSON format.
The Continents class contains two main methods. You're declaring the methods using the Python def keyword.
```
 class Continents:

     def get_db_connect(self):
     ...

     def get_continents(self):
     ...
```
The get_db_connect(self): method runs the mysql.connector.connect(...) function to connect to the MySQL server and returns a reusable connection using the return db_con statement.
The def get_continents(self): method constructs a query string (select * from continents) that retrieves records from the continents table. The dbCursor = dbConn.cursor(dictionary = True) statement instructs the MySQL driver for Python to format the database results as a dictionary. Then, the def get_continents(self): method returns the results in JSON format using the return json.dumps(continentsList , indent = 2) statement.
The Continents class is now ready for use and you can include and use it in other source code files using the following syntax:
```
 import continents
 c = continents.Continents()
 data = c.get_continents()
```
Continue to the next step and create a Redis® module to implement a rate-limiting feature in your application.

3. Create a `RateLimit` Class

This step takes you through coding a module that logs and remembers the number of requests a user makes to the API. Theoretically, you can implement the logging mechanism using a relational database. However, most relational databases are disk-based and may slow down your application. To avoid this slowness, an in-memory database like Redis® is more suitable.

The Redis® database server utilizes your computer's RAM for storage. RAM is several times faster than hard drives. To log users' requests, you need to create a unique key for each user in the Redis® server and set the key's expiry to a specific time (For instance, 60 seconds). Then, you must set the value of the Redis® key to the limit your API allows in that time frame. Then, when a user requests the API, decrement the key until their limits hit 0.

This sample API uses the users' HTTP Authorization request header as the Redis® key. The Authorization header is a Base64 encoded string containing a username and a password. For instance, the following is a Base64 encoded string for user john_doe who uses the password EXAMPLE_PASSWORD.

    +--------------------------+--------------------------------------+
    | Username:password        | Base64   encoded string              |
    +--------------------------+--------------------------------------+
    |john_doe:EXAMPLE_PASSWORD | am9obl9kb2U6RVhBTVBMRV9QQVNTV09SRA== |
    +--------------------------+--------------------------------------+

Because all users have different usernames, there is a guarantee that the Redis® key is unique. Execute the following steps to code the RateLimit class:

Open a new rate_limit.py file in a text editor.
```
 $ nano rate_limit.py
```

Enter the following information into the rate_limit.py file.

 import redis

 class RateLimit:

     def __init__(self):

         self.upperLimit = 5
         self.ttl = 60
         self.remainingTries = 0

     def get_limit(self, userToken):

         r = redis.Redis()

         if r.get(userToken) is None:

             self.remainingTries = self.upperLimit - 1

             r.setex(userToken, self.ttl, value = self.remainingTries)

         else:

             self.remainingTries = int(r.get(userToken).decode("utf-8")) - 1

             r.setex(userToken, r.ttl(userToken), value = self.remainingTries)

             self.ttl = r.ttl(userToken)

Save and close the rate_limit.py file.

The rate_limit.py file explained:

The import redis declaration imports the Redis® module for Python. This library provides Redis® functionalities inside the Python code.
The RateLimit class has two methods. You're defining these methods using the def __init__(self) and def get_limit(self, userToken) statements.
```
 class RateLimit:

     def __init__(self):
     ...

     def get_limit(self, userToken):

     ...
```
The def __init__(self): is a constructor method that initializes the default class members. The following list explains what each public class member does:
- self.upperLimit = 5: This variable sets the total number of requests allowed by the API in a time frame defined by the self.ttl variable.
- self.ttl = 60: This is a time frame in seconds during which the application watches the number of requests made by a particular API user. The ttl variable defines the Redis® key's time to live.
- self.remainingTries = 0: The API updates this value every time a user makes a request.
The def get_limit(self, userToken): method takes a userToken argument. This argument contains a Base64 encoded string containing the users' username and password (For instance, am9obl9kb2U6RVhBTVBMRV9QQVNTV09SRA==).
The r = redis.Redis() statement initializes the Redis® module for Python.
The logical if...else... statement queries the Redis® server (r.get(userToken)) to check if a key exists in the Redis® server matching the value of the userToken. If the key doesn't exist (during a new user's request), you're creating a new Redis® key using the r.setex(userToken, self.ttl, value = self.remainingTries) statement. If the key exists (during a repeat request), you're querying the value of the Redis® key using the int(r.get(userToken).decode("utf-8")) statement to update the remainingTries value. The r.setex(userToken, r.ttl(userToken), value = self.remainingTries - 1) statement allows you to decrement the Redis® key-value during each request.
```
         ...
         if r.get(userToken) is None:

             self.remainingTries = self.upperLimit - 1

             r.setex(userToken, self.ttl, value = self.remainingTries)

         else:

             self.remainingTries = int(r.get(userToken).decode("utf-8")) - 1

             r.setex(userToken, r.ttl(userToken), value = self.remainingTries)

             self.ttl = r.ttl(userToken)

         ...
```
The RateLimit class is now ready. You can include the class module in other source code files using the following declarations:
```
 import rate_limit
 rateLimit = rate_limit.RateLimit()
 rateLimit.get_limit(userToken)
```
Proceed to the next step to create an application's entry point.

4. Create a `main.py` File

You must define the main file that executes when you start your Python API. Follow the steps below to create the file.

Open a new main.py file in a text editor.
```
 $ nano main.py
```

Enter the following information into the main.py file.

 import http.server
 from http import HTTPStatus

 import socketserver
 import json

 import continents
 import rate_limit

 class httpHandler(http.server.SimpleHTTPRequestHandler):

     def do_GET(self):

         authHeader = self.headers.get('Authorization').split(' ');

         self.send_response(HTTPStatus.OK)
         self.send_header('Content-type', 'application/json')
         self.end_headers()

         rateLimit = rate_limit.RateLimit()
         rateLimit.get_limit(authHeader[1])

         ttl = rateLimit.ttl
         remainingTries = int(rateLimit.remainingTries)

         if remainingTries + 1 < 1:

            data = "You're not allowed to access this resource. Wait " + str(ttl) + " seconds before trying again.\r\n"

         else:

            c = continents.Continents()
            data = c.get_continents()

         self.wfile.write(bytes(data, "utf8"))

 httpd = socketserver.TCPServer(('', 8080), httpHandler)
 print("HTTP server started...")

 try:
     httpd.serve_forever()
 except KeyboardInterrupt:
     httpd.server_close()
     print("The server is stopped.")

Save and close the main.py file.

The main.py file explained:

The import... section imports all the modules required by your application. That is the HTTP server module (http.server), the network requests module (socketserver), the JSON formatting module (json), and the two custom modules you coded earlier (continents and rate-limit).
```
 import http.server
 from http import HTTPStatus

 import socketserver
 import json

 import continents
 import rate_limit
 ...
```

The httpHandler class handles the API HTTP requests.

 class httpHandler(http.server.SimpleHTTPRequestHandler):

    def do_GET(self):
        ...

The def do_GET(self): method runs when a user sends a request to the API using the HTTP GET request, as shown below.
```
 $ curl -X GET -u john_doe:EXAMPLE_PASSWORD  http://localhost:8080/
```
The authHeader = self.headers.get('Authorization').split(' '); statement retrieves a Base64 encoded string containing the API user's username and password.

The following declarations set the HTTP 200 status with the correct headers to allow HTTP clients to format the JSON output.

        ...
        self.send_response(HTTPStatus.OK)
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        ...

The code block below initializes the RateLimit class. After a user requests the API, you're querying the RateLimit class to check the user's limit using the rateLimit.get_limit(authHeader[1]) statement. Then, you're retrieving the value of the ttl and remainingTries variables using the rateLimit.ttl and rateLimit.remainingTries public class variables.
```
        ...
        rateLimit = rate_limit.RateLimit()
        rateLimit.get_limit(authHeader[1])

        ttl = rateLimit.ttl
        remainingTries = int(rateLimit.remainingTries)
        ...
```

The following source code sets an error message to the API user if they've hit their limit. Otherwise, the application grants access to the continents resource using the c = continents.Continents() and data = c.get_continents() statements.

        ...
        if remainingTries + 1 < 1:

           data = "You're not allowed to access this resource. Wait " + str(ttl) + " seconds before trying again.\r\n"

        else:

           c = continents.Continents()
           data = c.get_continents()

        self.wfile.write(bytes(data, "utf8"))
        ...

Towards the end of the main.py file, you're starting a web server that listens for incoming HTTP requests on port 8080.

 ...

 httpd = socketserver.TCPServer(('', 8080), httpHandler)
 print("HTTP server started...)

 try:
     httpd.serve_forever()
 except KeyboardInterrupt:
     httpd.server_close()
     print("The server is stopped.")

The main.py file is ready. Test the application in the next step.

5. Test the Application

You now have all the source code files required to run and test your API and the rate-limiter module. Execute the following steps to download the modules you've included in the application and run tests using the Linux curl command.

Install pip, a tool for installing Python modules/libraries.
```
 $ sudo apt update
 $ sudo apt install -y python3-pip
```

Use the pip module to install MySQL connector and Redis® modules for Python.

mysql-connector-python module:

  $ pip install mysql-connector-python

Output.

  ...
  Successfully installed mysql-connector-python-8.0.30 protobuf-3.20.1

redis module:

  $ pip install redis

Output.

  ...
  Successfully installed async-timeout-4.0.2 deprecated-1.2.13 packaging-2rsing-3.0.9 redis-4.3.4 wrapt-1.14.1

Run the application. Remember, your application's entry point is the main.py file.
```
 $ python3 main.py
```
Python starts a web server and displays the following output. The previous command has a blocking function. Don't enter any other command on your active terminal window.
```
 HTTP server started...
```
Establish another SSH connection to your server and run the following curl command. The [1-10] parameter value at the end of the URL allows you to send 10 queries to the API.
```
 $ curl -X GET -u john_doe:EXAMPLE_PASSWORD http://localhost:8080/?[1-10]
```

Confirm the output below. The API limits access to the application after 5 requests and also displays the error You're not allowed to access this resource. Wait 60 seconds before trying again..

 [1/10]: http://localhost:8080/?1 --> <stdout>
 --_curl_--http://localhost:8080/?1
 [
   {
     "continent_id": 1,
     "continent_name": "ASIA"
   },
   {
     "continent_id": 2,
     "continent_name": "AFRICA"
   },
   {
     "continent_id": 3,
     "continent_name": "NORTH AMERICA"
   },
   {
     "continent_id": 4,
     "continent_name": "SOUTH AMERICA"
   },
   {
     "continent_id": 5,
     "continent_name": "ANTARCTICA"
   },
   {
     "continent_id": 6,
     "continent_name": "EUROPE"
   },
   {
     "continent_id": 7,
     "continent_name": "AUSTRALIA"
   }
 ]
 ...
 [5/10]: http://localhost:8080/?5 --> <stdout>
 --_curl_--http://localhost:8080/?5
 [
   {
     "continent_id": 1,
     "continent_name": "ASIA"
   },
   {
     "continent_id": 2,
     "continent_name": "AFRICA"
   },
   {
     "continent_id": 3,
     "continent_name": "NORTH AMERICA"
   },
   {
     "continent_id": 4,
     "continent_name": "SOUTH AMERICA"
   },
   {
     "continent_id": 5,
     "continent_name": "ANTARCTICA"
   },
   {
     "continent_id": 6,
     "continent_name": "EUROPE"
   },
   {
     "continent_id": 7,
     "continent_name": "AUSTRALIA"
   }
 ]
 [6/10]: http://localhost:8080/?6 --> <stdout>
 --_curl_--http://localhost:8080/?6
 You're not allowed to access this resource. Wait 60 seconds before trying again.

 [7/10]: http://localhost:8080/?7 --> <stdout>
 --_curl_--http://localhost:8080/?7
 ...

Conclusion

This guide shows you how to code a rate-limiter for a Python API with Redis® on Ubuntu 20.04. The API in this guide allows 5 requests in a 60-second time frame. However, you can adjust the RateLimit class self.upperLimit = 5 and self.ttl = 60 values based on the users' workloads and your application's capability.

Rate Limit a Python API with Redis®

Introduction

Prerequisites

1. Set Up a Sample Database and User Account

2. Create a `Continents` Class

3. Create a `RateLimit` Class

4. Create a `main.py` File

5. Test the Application

Conclusion

Products

Features

Marketplace

Resources

Company

Rate Limit a Python API with Redis®

Introduction

Prerequisites

1. Set Up a Sample Database and User Account

2. Create a Continents Class

3. Create a RateLimit Class

4. Create a main.py File

5. Test the Application

Conclusion

2. Create a `Continents` Class

3. Create a `RateLimit` Class

4. Create a `main.py` File