How to Use Redis with Python

When it comes to data-driven applications, developers and data engineers are always trying to balance factors such as scalability, speed, flexibility, latency, and availability.

In other words, databases and infrastructure are the foundations for well-structured applications: just like bricks are for houses.

This article explores Redis' data store features and includes use cases. We'll learn how to use Redis in Python with a step-by-step tutorial.

Let's get started!

What Is Redis?

Redis (short for Remote Dictionary Server) is an open-source, in-memory data structure store that can be used as a database, cache, message broker, or queue. It is known for its high performance, low latency, and versatility as it stores data in memory (on the RAM) rather than on disk, which makes it extremely fast for read and write operations.

Redis is often referred to as a data structure server because it provides access to mutable data structures via a set of commands, which are sent using a server-client model with TCP sockets and a simple protocol. This means that different processes can query and modify the same data structures in a shared way.

Why Use Redis?

Before describing Redis' most common use cases, let's explore its main features:

In-memory storage: In Redis, data is stored in RAM, providing response times in sub-milliseconds.
Persistence options: While data is stored in memory, Redis also offers persistance options like RDB (Redis Database Backup) — periodic dataset snapshots — and AOF (Append-Only File), which logs every write operation for durability. This allows data to be saved to disk and replicated across multiple servers to ensure data durability and availability.
Pub/sub messaging: It supports publish/subscribe messaging patterns for real-time communication.
Clustering and replication: Redis supports clustering and primary-secondary replication, providing scalability and high availability.

Finally, at its core, Redis is a key-value store, meaning it stores data as a collection of keys and their associated values.

A key in Redis is a unique string that acts as an identifier for the associated value. Keys are case-sensitive and can contain any binary sequence (e.g., strings, numbers, or special characters).

A value in Redis can be one of several data types:

Strings
Hashes
Lists
Sets
Sorted Sets
Bitmaps
HyperLogLogs
Streams

Redis provides a wide set of commands for working with key-value pairs, such as SET, GET, and DEL for strings, HSET, HGET, and HDEL for hashes, and LPUSH, LGET, and LREM for lists. Here are some examples of key-value pairs:

SET user:1001 "John Doe"
HSET user:1001 name "John Doe" age 30 email "john@example.com"
LPUSH tasks "task1" "task2" "task3"

The management of such data structures in Redis is important because:

Redis stores them on disk, even if they are always served and modified into server memory. This means that Redis is fast, but also non-volatile.
The implementation of data structures emphasizes memory efficiency, so data structures inside Redis will likely use less memory compared to the same data structure modelled using a high-level programming language.
Redis offers a number of features commonly found in databases, such as replication, tunable durability levels, clustering, and high availability.

With this in mind, here are some common applications of Redis:

Caching: As it stores data in memory, one of the main uses of Redis is as a caching system for web applications, since it reduces server load and improves page loading times.

How Redis caches web requests by Federico Trotta

Real-time applications: Given its speed and reliability, Redis is perfect for real-time use cases like chat applications, real-time analytics, and live notifications.
Session management: Another common use of Redis is storing user session data in web applications because of its speed and ability to handle expiration policies.
Message queues: Redis can act as a lightweight message broker using its pub/sub or streams features, making it suitable for event-driven architectures.
Advanced data structures management: Redis supports advanced data structures that are not available in traditional databases, making it useful for tasks like counting unique visitors (HyperLogLog), storing time-series data (Streams), and implementing priority queues (Sorted Sets).

How to Use Redis in Python: A Step-By-Step Tutorial

Now that we've covered the theory, it's time to see how to use Redis in Python.

Prerequisites

To replicate this tutorial, your system must have the following prerequisites:

Python 3.6 or higher: Any Python version higher than 3.6 will do. Specifically, we will install dependencies via pip (already included in any Python version from 3.4+).

Step 1: Setting Up the Environment and Installing Dependencies

Suppose you call the main folder of your project redis. The project structure will be as follows:

plaintext

redis/
│
├── venv/
└── examples/
   ├── basic_connection.py
   ├── strings_example.py
   ├── hashes_example.py
   ├── sets_example.py
   ├── transactions_example.py
   └── expiration.py

Where:

The .py files will contain the logic described in the next steps.
venv/ contains the virtual environment.

You can create the venv/ virtual environment directory like so:

Shell

python -m venv

To activate it on Windows, run:

Shell

venv\Scripts\activate

On macOS/Linux, execute:

Shell

source venv/bin/activate

In the activated virtual environment, install Redis:

Shell

pip install redis

Step 2: Testing The Connection to Redis

To establish a connection to a Redis server using redis-py, type the following code into the basic_connection.py file:

Python

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
 
print(r.ping())

When you run it via python3 basic_connection.py, if everything works fine, you will receive:

Shell

True

Step 3: Managing Data Structures

We've learned that Redis stores data as key-value pairs, and the values can be different data structures. Now it's time to see how this works.

To manage strings, type the following in strings_example.py:

Python

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
 
r.set('user:1001', 'John Doe')
print(r.get('user:1001'))

Which, as expected, returns:

Shell

John Doe

To manage hashes, type the following in the hashes_example.py file:

Python

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
 
r.hset('user:1', 'name', 'Alice')
print(r.hget('user:1', 'name'))

Which returns:

Shell

Alice

To manage sets, write the following into the sets_example.py file:

Python

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
 
r.sadd('tags', 'python', 'redis')
print(r.smembers('tags'))

Which returns:

Shell

{'redis', 'python'}

Step 4: Using Advanced Redis Features

Redis provides various advanced features, including pipelines.

A pipeline in Redis is a way to batch multiple commands together and send them to the Redis server in a single network request. This reduces the overhead of multiple round trips between the client and the server, improving performance, especially when executing multiple commands.

For example, if you want to create a pipeline object (pipe) that queues commands instead of executing them immediately, type the following into the transactions_example.py file:

Python

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
 
with r.pipeline() as pipe:
    pipe.set('key1', 'value1')
    pipe.set('key2', 'value2')
    pipe.execute()

The pipe.execute() method sends all the queued commands in the pipeline to the Redis server in a single request. Then, the server processes the commands in the order they are queued.

Another advanced feature of Redis is the possibility to set keys with expiration times, retrieve them before and after expiration, and handle scenarios where the keys no longer exist. To do so, write the following into the expiration.py file:

Python

import time
import redis
 
# Connect to the Redis server
r = redis.Redis(host='localhost', port=6379, db=0)
 
# Set keys with expiration using setex()
print("Setting keys with expiration...")
r.setex('temp_key1', 5, 'value1')  # Expires in 5 seconds
r.setex('temp_key2', 10, 'value2')  # Expires in 10 seconds
 
# Retrieve the keys before expiration
print("\nRetrieving keys before expiration:")
print(f"temp_key1: {r.get('temp_key1')}")
print(f"temp_key2: {r.get('temp_key2')}")
 
# Wait for 6 seconds
print("\nWaiting for 6 seconds...")
time.sleep(6)
 
# Try retrieving the keys after expiration
print("\nRetrieving keys after 6 seconds:")
print(f"temp_key1: {r.get('temp_key1')}")
print(f"temp_key2: {r.get('temp_key2')}")
 
# Wait for another 5 seconds
print("\nWaiting for another 5 seconds...")
time.sleep(5)
 
# Try retrieving the keys again
print("\nRetrieving keys after 11 seconds:")
print(f"temp_key1: {r.get('temp_key1')}")
print(f"temp_key2: {r.get('temp_key2')}")
 
# Set a key with expiration and check its TTL
print("\nSetting a new key with expiration...")
r.setex('temp_key3', 15, 'value3')
print(f"temp_key3: {r.get('temp_key3')}")
print(f"TTL for temp_key3: {r.ttl('temp_key3')} seconds")
 
# Wait for 5 seconds and check TTL again
print("\nWaiting for 5 seconds...")
time.sleep(5)
print(f"TTL for temp_key3 after 5 seconds: {r.ttl('temp_key3')} seconds")
 
# Delete the key before it expires
print("\nDeleting temp_key3 before it expires...")
r.delete('temp_key3')
print(f"temp_key3: {r.get('temp_key3')}")

Here is what this code does:

The setex(name, time, value) method sets a key with a specific expiration time (in seconds). In the provided example, the keys have different expiration times.
The get() method retrieves the value of a key. Before the expiration time, the keys return their respective values.
The ttl() method retrieves the remaining time (in seconds) before a key expires.

Here is the expected result after the whole process is completed:

Shell

Setting keys with expiration...
 
Retrieving keys before expiration:
temp_key1: b'value1'
temp_key2: b'value2'
 
Waiting for 6 seconds...
 
Retrieving keys after 6 seconds:
temp_key1: None
temp_key2: b'value2'
 
Waiting for another 5 seconds...
 
Retrieving keys after 11 seconds:
temp_key1: None
temp_key2: None
 
Setting a new key with expiration...
temp_key3: b'value3'
TTL for temp_key3: 15 seconds
 
Waiting for 5 seconds...
TTL for temp_key3 after 5 seconds: 10 seconds
 
Deleting temp_key3 before it expires...
temp_key3: None

How to Use Redis in Python for Monitoring

Monitoring a Redis server is an essential aspect of managing a production system, for more than just checking general statistics.

Redis offers built-in methods for monitoring, like:

info(): Providing comprehensive details about memory usage, client connections, CPU utilization, and other important statistics.
monitor(): Streaming every command processed by the Redis server in real time. (However, consider that using this in a production environment can be resource-intensive and may impact performance.)
client_list(): Returning details about connected clients, which is useful for identifying unusual client behavior or ensuring your connection pools are behaving as expected.
slowlog_get(): Allowing you to track slow-executing commands to pinpoint potential bottlenecks or inefficient queries.

Here's how to implement the above monitoring methods:

Python

import redis
import threading
import time
 
def server_stats_monitor():
    """
    Periodically retrieves and prints key Redis server statistics.
    It uses INFO to extract memory usage, client counts, CPU usage, and uptime.
    """
    r = redis.Redis(host='localhost', port=6379, db=0)
 
    while True:
        info = r.info()
        print("=== Redis Server Monitoring ===")
        print("Used Memory:", info.get("used_memory_human"))
        print("Connected Clients:", info.get("connected_clients"))
        print("CPU Usage:", info.get("used_cpu_sys"), "system CPU seconds,",
              info.get("used_cpu_user"), "user CPU seconds")
        print("Uptime (seconds):", info.get("uptime_in_seconds"))
        print()  # Blank line for readability
 
        # Retrieve and display slow log entries (if any)
        slow_logs = r.slowlog_get()
        print("=== Slow Log Entries ===")
        if slow_logs:
            for log in slow_logs:
                # Each log entry is a dictionary with keys like 'id', 'start_time', 'duration', and 'command'
                print(f"ID: {log.get('id')}, Duration: {log.get('duration')} μs, Command: {log.get('command')}")
        else:
            print("No slow log entries found.")
        print()
 
        # Retrieve and display client list details
        client_list = r.client_list()
        print("=== Connected Clients ===")
        for client in client_list:
            # Displaying each client info (address, age, idle time, flags, db, etc.)
            print(f"Address: {client.get('addr')} | Idle Time: {client.get('idle')} sec | DB: {client.get('db')}")
        print("\n-----------------------------\n")
 
        time.sleep(10)  # Pause for 10 seconds before repeating
 
# Start the monitoring in a background thread so that it runs continuously.
stats_thread = threading.Thread(target=server_stats_monitor)
stats_thread.daemon = True
stats_thread.start()
 
 
def monitor_commands():
    """
    Sets up a real-time monitor that listens to every command processed by Redis.
    Note: Running MONITOR can be heavy on performance and should be used cautiously.
    """
    r = redis.Redis(host='localhost', port=6379, db=0)
    monitor = r.monitor()
    for command in monitor.listen():
        print("Real-time Command:", command)
 
 
# Keep the main thread active so the background monitoring continues.
while True:
    time.sleep(1)

This code does the following:

It prints server statistics every ten seconds using the info() method.
slowlog_get() gets slow log entries, if any. Each entry is typically a dictionary containing an identifier, start time, duration (in microseconds), and the command that ran slowly. This helps you to identify any commands that might be causing performance issues.
The client_list() gets the connected clients' status. The returned result is a list of dictionaries, each containing details such as the client's address, idle time, connected database, and other flags. Monitoring these details can help you find issues with idle or misbehaving clients.
An optional real-time monitor using monitor() is provided to get real time data.

The expected result is something like this:

Shell

=== Redis Server Monitoring ===
Used Memory: 852.59Kz\
Connected Clients: 1
CPU Usage: 2.931738 system CPU seconds, 1.491924 user CPU seconds
Uptime (seconds): 1773
 
=== Slow Log Entries ===
No slow log entries found.
 
=== Connected Clients ===
Address: xxx.x.x.x:xxxxx | Idle Time: 0 sec | DB: 0

But, of course, it will continue to provide statistics in a loop:

Monitoring Redis with AppSignal

While Redis provides built-in monitoring features, they are low-level, infrastructure-focused metrics for understanding raw performance, resource consumption, and debugging specific issues within the Redis server itself.

If you want to develop a more detailed understanding of your application's overall health, you can use AppSignal's integration with Redis.

AppSignal provides performance metrics, error tracking, and logging for your applications.

To use AppSignal, first sign up for a 30-day free trial, use the Python installation guide and configuration guide.

Then you can integrate AppSignal with Redis like so:

Python

import redis
import threading
import time
from appsignal import Appsignal
 
# Initialize the AppSignal client with your AppSignal push API key and application name.
appsignal_instance = Appsignal(
    push_api_key="YOUR_APPSIGNAL_PUSH_API_KEY",
    name="redis_monitor",
    active=True
)
 
def monitor_redis_with_appsignal():
    """
    This function collects Redis monitoring data and sends selected metrics
    to AppSignal using its Python API. Metrics include memory usage, client counts,
    uptime, slow log details, and more.
    """
    # Connect to your Redis server
    r = redis.Redis(host='localhost', port=6379, db=0)
 
    while True:
        # Collect basic server stats using the INFO command
        info = r.info()
        metrics = {
            "redis.used_memory_bytes": info.get("used_memory"),
            "redis.used_memory_human": info.get("used_memory_human"),
            "redis.connected_clients": info.get("connected_clients"),
            "redis.cpu_sys": info.get("used_cpu_sys"),
            "redis.cpu_user": info.get("used_cpu_user"),
            "redis.uptime_seconds": info.get("uptime_in_seconds")
        }
 
        # Retrieve slow log entries and add their count and maximum duration (if any)
        slow_logs = r.slowlog_get()
        if slow_logs:
            metrics["redis.slowlog_count"] = len(slow_logs)
            max_duration = max(log.get("duration", 0) for log in slow_logs)
            metrics["redis.max_slowlog_duration_microseconds"] = max_duration
        else:
            metrics["redis.slowlog_count"] = 0
 
        # Retrieve client list details
        client_list = r.client_list()
        metrics["redis.client_count_detail"] = len(client_list)
 
        # Use the AppSignal API to record each metric.
        # Here we assume appsignal_instance.record_metric() takes a metric name and a value.
        for metric_name, value in metrics.items():
            appsignal_instance.record_metric(metric_name, value)
 
        # Log the transmission (for debugging/demonstration purposes)
        print("Sent the following metrics to AppSignal:")
        for k, v in metrics.items():
            print(f"  {k}: {v}")
        print("\n---\n")
 
        # Wait for 10 seconds before collecting metrics again
        time.sleep(10)
 
# Start the monitoring process in a background thread
monitor_thread = threading.Thread(target=monitor_redis_with_appsignal)
monitor_thread.daemon = True
monitor_thread.start()
 
# Keep the main program alive so that the background monitoring continues indefinitely.
while True:
    time.sleep(1)

This example instantiates the Appsignal() class to send Redis monitoring metrics directly to AppSignal. This approach allows you to:

Correlate Redis data with transaction traces and error reports from the rest of your application. For example, if a spike in redis.slowlog_count coincides with a surge in request latency or error notifications, you have immediate insight into a potential bottleneck or failure point.
Visualize metric history over time, as AppSignal stores historical data. This allows you to identify trends, seasonal variations, or gradual degradations in Redis performance — insights that are not as obvious when only using Redis’s native monitoring features.
Set up custom alerts in AppSignal that are triggered when certain thresholds are breached.

Here is an example of an error you might see in AppSignal's Errors dashboard: