Python Redis: A Beginner's Guide

Using Redis with Python is a powerful combination for data engineers looking to build high-performance, scalable, and maintainable systems. Whether you're optimizing caching layers, managing real-time event streams, or implementing distributed locking, Redis provides a flexible and efficient solution.

In this guide, we’ll cover the basics of using Redis with Python to help you integrate it seamlessly into your data engineering workflows.

What is Python Redis?

Redis is an open-source, in-memory data structure store that is used as a database, cache, and message broker. It stands for Remote Dictionary Server and was originally designed to be a key-value storage system with fast access times.

It supports various data structures such as strings, hashes, lists, sets and sorted sets with range queries. Redis also provides advanced features like transactions, pub/sub messaging, Lua scripting, and built-in replication and clustering capabilities for better availability and scalability.

Common Redis use cases

The potential use cases for Redis are vast and varied.

Some of the most common use cases in data engineering include: 

• Caching: Accelerate database queries and API responses.
• Message queues: Power event-driven architectures and job processing pipelines.
• Real-time analytics: Process and analyze streaming data with minimal latency.
• Leaderboards: Efficiently rank and retrieve top-scoring entities.
• Session storage: Maintain fast, scalable session state management.
• Job queue management: Ensure efficient task scheduling and execution.
• Geospatial Indexing: Store and query location-based data.

Redis is also commonly used in combination with other technologies such as web servers (e.g. NGINX), databases (e.g. MySQL), and messaging systems (e.g. Kafka) to optimize performance and scalability.

Why use Redis with Python?

Python is a dominant language in data engineering due to its flexibility, extensive ecosystem, and ease of integration with big data technologies. 

Combining it with Redis can bring many benefits to your applications:

• Fast performance: As Redis stores data in memory, it can offer significantly faster access times compared to traditional databases which store data on disk. This makes it ideal for use cases that require high performance, such as caching and real-time applications.
• Scalability: Redis is designed with scalability in mind, making it easy to handle large amounts of data without compromising performance.
• Ease of use: Python's simple syntax and rich library ecosystem make it easy to work with Redis. The official Redis-py client library also provides a user-friendly interface for interacting with Redis from within your Python code.

What is Redis-py?

Redis-py is the official Redis client library for Python. It provides a user-friendly interface for communicating with Redis from within your Python code.

Some of its main features include:

• Support for all Redis data types
• Pipeline and transaction support for efficient and atomic operations
• Pub/sub messaging support
• Automatic encoding/decoding of data to/from bytes, strings, and other Python types
• Cluster mode support for working with Redis clusters

Setting Up Redis for Python Development

Let's have a look at how we can set up Redis for ourselves using Python.

There are three main ways to get Redis running:

1. Docker (Recommended for data engineers)
2. Windows Subsystem for Linux (WSL)
3. Third-Party Builds (e.g., Memurai, tporadowski/redis)

Method A: Using Docker

Firstly, when using Docker, you’ll need to have Docker desktop installed in your machine. This can be downloaded from the DockerHub website.

Once Docker is installed, open up your command prompt and key in the following command.

docker pull redis

Next, we’ll start a new container.

docker run --name my-redis -p 6379:6379 -d redis

docker pull redis downloads the latest official Redis image. docker run --name my-redis -p 6379:6379 -d redis starts a new container named my-redis. The -p 6379:6379 parameter maps your local machine’s port 6379 to the container’s port 6379, so you can connect to Redis at localhost:6379. The -d flag runs the container in the background (detached mode).

To verify, run:

docker exec -it my-redis redis-cli ping

It should respond with PONG.

Method B: Using Windows Subsystem for Linux (WSL)

For Windows machines, another possible method is to use Windows Subsystem for installation. This will use Linux to do the installation.

wsl --install

This command installs and configures the default Ubuntu distribution inside WSL. Reboot if required.

Then, open your Ubuntu terminal (from Microsoft Store or by typing wsl in PowerShell) and run:

sudo apt-get update
sudo apt-get install redis-server

Next, start your Redis server.

sudo service redis-server start

Let’s do a test to see if the server is up and running.

redis-cli ping

Here’s a detailed description of the commands we used above:

• sudo apt-get update fetches the latest package definitions.
• sudo apt-get install redis-server installs Redis within the Ubuntu environment.
• sudo service redis-server start immediately starts the Redis service.
• redis-cli ping tests connectivity to the local Redis server, returning PONG if successful.

Method C: Third-party builds

• Memurai: Memurai.com
• tporadowski/redis: GitHub releases

These solutions provide native Windows binaries. After downloading and installing or unzipping, run:

redis-server.exe

Once launched, redis-server.exe should print logs indicating it is listening on 127.0.0.1:6379.

You can confirm by using the included CLI (if provided):

redis-cli.exe ping

You should expect a PONG result if successful.

Installing the redis-py Library

Next, you’ll need to install the redis-py library for you to access the database using Python. Before you install, the library, make sure you have both Python and Redis installed separately.

pip install redis

This command downloads and installs the Python Redis client library (redis-py) from PyPI. If you are using a virtual environment, make sure you activate it before running this command.

Verify using:

python -c "import redis; print(redis.__version__)"

This should print a version number (e.g., 5.2.1).

Setting up a Redis server

If you followed any method above (Docker, WSL, or a third-party build), you should now have a Redis server running on localhost:6379. That’s sufficient for local development on Windows.

Working with Redis Data Structures

Below are Python examples demonstrating how to connect to and manipulate different Redis data types. Each code block includes explanatory comments detailing how the methods or commands function.

Strings in Redis

import redis

# Instantiate a Redis client, connecting to localhost on port 6379
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0  # The default Redis database index
)

# 1. SET command: store a string under 'mykey'
r.set("mykey", "hello from Windows")

# 2. GET command: retrieve the value stored at 'mykey'
value = r.get("mykey")
print(value)   # Output is b'hello from Windows', since redis-py returns bytes.

# 3. Convert bytes to string
print(value.decode())  # prints "hello from Windows"

# 4. DEL command: remove 'mykey' from Redis
r.delete("mykey")

We create a Redis object that connects to the local Redis server.

• r.set("mykey", "hello from Windows") stores the string "hello from Windows" under the key "mykey".
• r.get("mykey") fetches the value stored in mykey. Redis returns bytes, so we decode it for human-readable output.
• r.delete("mykey") deletes the key mykey altogether.

Example: Storing and retrieving user sessions

Here’s a code example of user session management:

import uuid

def store_user_session(user_id):
    """
    Generates a unique session token (UUID) for a user and stores it in Redis.
    The session is stored under the key pattern: user:{user_id}:session
    """
    session_key = f"user:{user_id}:session"
    token = str(uuid.uuid4())   # Generate a random UUID as a session token
    r.set(session_key, token)   # Store token in Redis
    return token

def get_user_session(user_id):
    """
    Retrieves the stored session token for the given user_id.
    Returns None if the session does not exist or is expired.
    """
    session_key = f"user:{user_id}:session"
    token = r.get(session_key)
    return token.decode('utf-8') if token else None

def delete_user_session(user_id):
    """
    Deletes the session entry from Redis for the specified user_id.
    """
    session_key = f"user:{user_id}:session"
    r.delete(session_key)

# Usage demonstration
session_token = store_user_session(1001)
print(f"Stored session token: {session_token}")

retrieved_token = get_user_session(1001)
print(f"Retrieved session token: {retrieved_token}")

delete_user_session(1001)
print(f"Session after delete: {get_user_session(1001)}")  # Should be None or empty

store_user_session(user_id): Creates a globally unique ID via uuid.uuid4() and stores it under a user-specific key. get_user_session(user_id): Reads back the session token. If None, the session is missing or expired. delete_user_session(user_id): Removes the session key from Redis, effectively logging out the user.

Lists in Redis

Redis Lists are ordered sequences of strings. They allow operations on both ends, making them handy for queues, stacks, or logs.

# LPUSH: Push an element to the head (left) of the list
r.lpush("task_queue", "task1")

# RPUSH: Push an element to the tail (right) of the list
r.rpush("task_queue", "task2")
r.rpush("task_queue", "task3")

# LPOP: Pop (remove and return) the element at the head
task = r.lpop("task_queue")
print(task)  # b'task1'

# Optional: RPOP removes and returns the element at the tail
task = r.rpop("task_queue")
print(task)  # b'task3'

lpush("task_queue", "task1") adds "task1" to the left side of a list called "task_queue". rpush("task_queue", "task2") and rpush("task_queue", "task3") add tasks to the right side. lpop("task_queue") and rpop("task_queue") remove and return items from the respective ends of the list.

Example: Implementing a simple Redis-backed queue

def enqueue_task(queue_name, task):
    """
    Appends a task to the end (right) of the Redis list named queue_name.
    """
    r.rpush(queue_name, task)

def dequeue_task(queue_name):
    """
    Removes a task from the front (left) of the Redis list named queue_name.
    Returns the task as a string, or None if the queue is empty.
    """
    task = r.lpop(queue_name)
    return task.decode('utf-8') if task else None

# Example usage:
enqueue_task("my_queue", "send_email")
enqueue_task("my_queue", "generate_report")

while True:
    task = dequeue_task("my_queue")
    if not task:
        print("No more tasks in queue.")
        break
    print(f"Processing task: {task}")

enqueue_task always adds new tasks to the end of the list, which simulates a traditional FIFO queue. dequeue_task removes tasks from the front. If the list is empty, it returns None.

The while-loop processes tasks until the queue is empty, then stops.

Hashes in Redis

Hashes are similar to Python dictionaries but stored in Redis. They’re best for grouping related fields (e.g., user details).

# HSET: Store 'name' and 'email' fields for a user hash key
r.hset("user:1001", "name", "Alice")
r.hset("user:1001", "email", "alice@example.com")

# HGET: Retrieve a single field from the hash
email = r.hget("user:1001", "email")
print(email.decode('utf-8'))  # alice@example.com

# HDEL: Remove a field from the hash
r.hdel("user:1001", "email")

hset("user:1001", "name", "Alice") sets the field "name" to the value "Alice" within the hash at key "user:1001". hget("user:1001", "email") fetches the "email" field. hdel("user:1001", "email") deletes the "email" field from that hash.

Example: Storing and accessing structured user profiles

def create_user_profile(user_id, name, email):
    """
    Creates a user profile in Redis under the key 'user:{user_id}'.
    'name' and 'email' are stored as separate fields in the hash.
    """
    user_key = f"user:{user_id}"
    r.hset(user_key, mapping={"name": name, "email": email})

def get_user_profile(user_id):
    """
    Retrieves and returns all fields in the user profile hash
    as a Python dictionary. Keys and values are decoded from bytes.
    """
    user_key = f"user:{user_id}"
    profile_data = r.hgetall(user_key)
    return {k.decode('utf-8'): v.decode('utf-8') for k, v in profile_data.items()}

def delete_user_profile(user_id):
    """
    Deletes the entire user profile key from Redis.
    """
    user_key = f"user:{user_id}"
    r.delete(user_key)

# Usage demonstration
create_user_profile(1002, "Bob", "bob@example.com")
print(get_user_profile(1002))  # e.g. {'name': 'Bob', 'email': 'bob@example.com'}
delete_user_profile(1002)

We store a user’s name and email inside a hash. mapping={"name": name, "email": email} uses redis-py’s ability to set multiple fields at once. r.hgetall(user_key) retrieves all fields and values, returned as a dictionary of raw bytes which we decode to strings. delete_user_profile removes the entire hash key.

Sets and Sorted Sets

Data in Redis can be managed using sets or sorted sets. 

Sets are collections of unique values, while sorted sets are collections of key-value pairs with a score associated with each value. These data structures provide efficient ways to store and retrieve data, making them useful for applications that require fast lookups.

Sets

A set in Redis is similar to a mathematical set – it contains a collection of unique elements, and no duplicate elements are allowed. Sets can be used to model relationships between different entities, such as users who have liked a post on social media.

Here’s a demonstration of sets:

# SADD: Add multiple members to a set
r.sadd("tags:python", "redis", "windows", "backend")

# SMEMBERS: Retrieve all unique members in the set
tags = r.smembers("tags:python")
print(tags)  # {b'redis', b'windows', b'backend'}

Redis Sets ensure uniqueness of members. Attempting to add a duplicate member will have no effect.

The result of smembers is a Python set of bytes.

Sorted sets

Sorted sets in Redis are similar to regular sets, but each member also has a corresponding score. This allows for efficient sorting and ranking operations on the set.

Sorted sets are often used for leaderboards, where the score represents a player's rank or points.

Here’s a demonstration of sorted sets:

# ZADD: Add members with scores
r.zadd("leaderboard", {"player1": 10, "player2": 20})

# ZRANGE: Retrieve members in ascending order of score
leaders = r.zrange("leaderboard", 0, -1, withscores=True)
print(leaders)  # [(b'player1', 10.0), (b'player2', 20.0)]

Sorted sets store members in a specific order determined by their numeric scores. zrange("leaderboard", 0, -1, withscores=True) returns all members from rank 0 to the end, including their scores.

Example: Managing tags or leaderboards

The use of sorted sets can be helpful in managing tags or creating leaderboards. 

def add_tag(post_id, tag):
    """
    Adds a 'tag' to the set of tags belonging to a specific post.
    Each post has its own set under 'post:{post_id}:tags'.
    """
    r.sadd(f"post:{post_id}:tags", tag)

def get_tags(post_id):
    """
    Retrieves all tags for a specific post, decoding the bytes into strings.
    """
    raw_tags = r.smembers(f"post:{post_id}:tags")
    return {tag.decode('utf-8') for tag in raw_tags}

def update_leaderboard(player, score):
    """
    Updates or inserts a player's score in the 'game:leaderboard' sorted set.
    A higher score indicates a better position if sorting descending.
    """
    r.zadd("game:leaderboard", {player: score})

def get_leaderboard():
    """
    Returns an ascending list of (player, score) tuples from the leaderboard.
    To invert the sorting (highest first), you'd use ZREVRANGE.
    """
    entries = r.zrange("game:leaderboard", 0, -1, withscores=True)
    return [(player.decode('utf-8'), score) for player, score in entries]

# Usage demonstration
add_tag(123, "python")
add_tag(123, "redis")
print(get_tags(123))

update_leaderboard("Alice", 300)
update_leaderboard("Bob", 450)
print(get_leaderboard())

add_tag and get_tags demonstrate a one-to-many tagging system, storing tags in a Set. update_leaderboard and get_leaderboard show how you might implement a game leaderboard.

Redis Features in Python Applications

Redis has several features that make it a popular choice for data storage in Python applications: queueing, locking, and caching. 

Here are their implementations in Python below:

Redis as a queue (with blocking)

Let’s explore using Redis as a queue with blocking using BLPOP.

BLPOP is a blocking operation that waits until an element is available in a list.

def blocking_consumer(queue_name):
    """
    Continuously listens to the specified queue (Redis list) using BLPOP,
    which blocks until new items are pushed. Once an item arrives,
    it is removed from the queue and processed.
    """
    print(f"Waiting on queue: {queue_name}")
    while True:
        result = r.blpop(queue_name)
        if result:
            list_name, task_bytes = result
            task = task_bytes.decode('utf-8')
            print(f"Received task: {task}")
        else:
            print("Queue is empty or an error occurred.")
            break

def enqueue_task(queue_name, task):
    """
    Pushes a task to the end of a Redis list (queue).
    """
    r.rpush(queue_name, task)

# Example usage:
enqueue_task("blocking_queue", "task_block_1")
enqueue_task("blocking_queue", "task_block_2")

# In a real application, the consumer might run in a separate thread or process
blocking_consumer("blocking_queue")

blocking_consumer uses blpop in a loop. If the list is empty, blpop will wait until another item is pushed.

Once an item is received, it’s removed from the list and printed.

This approach is ideal for producer-consumer patterns where workers consume tasks as they appear.

Implementing locks in Redis

Distributed locks prevent multiple clients or processes from simultaneously modifying the same resource. Redis can help avoid race conditions in a distributed environment.

import time
from redis.exceptions import LockError

def process_critical_section():
    """
    Acquires a lock named 'resource_lock' with a timeout of 10 seconds.
    The lock automatically expires after 10 seconds to prevent deadlocks.
    """
    lock = r.lock("resource_lock", timeout=10)
    try:
        # Attempt to acquire the lock, wait for up to 5 seconds if another process holds it
        acquired = lock.acquire(blocking=True, blocking_timeout=5)
        if acquired:
            print("Lock acquired; performing critical operation...")
            time.sleep(3)  # Simulate some operation
        else:
            print("Failed to acquire lock within 5 seconds.")
    except LockError:
        print("A LockError occurred, possibly releasing already released lock.")
    finally:
        # Always release the lock in a finally block to ensure cleanup
        lock.release()
        print("Lock released.")

# Usage demonstration
process_critical_section()

We create a lock with timeout=10, meaning if the lock isn’t released manually, Redis will automatically remove it in 10 seconds to prevent indefinite blocking (deadlock). lock.acquire(blocking=True, blocking_timeout=5) tries to acquire the lock for 5 seconds before giving up. After finishing, lock.release() frees the resource for other processes.

Caching with Redis

Caching is a common usage scenario: store frequently accessed data in memory, reducing load on databases or external APIs.

import requests
import json

def get_user_data(user_id):
    """
    Retrieves user data from a hypothetical API endpoint.
    If the data is found in Redis (cache), use that. Otherwise, call the API,
    store the response in Redis with a 60-second expiration, and return it.
    """
    cache_key = f"user_data:{user_id}"
    cached_data = r.get(cache_key)
    if cached_data:
        print("Cache hit!")
        return json.loads(cached_data)

    print("Cache miss. Fetching from API...")
    response = requests.get(f"https://api.example.com/users/{user_id}")
    user_info = response.json()

    # Store in Redis for 60 seconds
    r.setex(cache_key, 60, json.dumps(user_info))
    return user_info

# Usage
user = get_user_data(42)  # First call => cache miss
user_again = get_user_data(42)  # Subsequent call => cache hit

We check if user_data:{user_id} exists in Redis. If it does, that’s a cache hit, and we skip the API call. If not, we fetch from the remote API, then setex (set + expiration) for 60 seconds. Subsequent calls within that timeframe will retrieve the cached data, reducing latency.

Advanced Topics in Redis

Think you’ve got the hang of Redis already? Let’s go through some more advanced topics.

Using Pub/Sub with Redis

Pub/Sub (Publish/Subscribe) is a popular messaging pattern for real-time communication. Publishers send messages to a channel, and all subscribers to that channel receive the messages.

import threading

def subscriber(r, channel_name):
    """
    Subscribes to the given Redis channel and listens for messages.
    When a new message is published on that channel, it is printed.
    """
    pubsub = r.pubsub()
    pubsub.subscribe(channel_name)
    print(f"Subscribed to {channel_name}")

    # pubsub.listen() yields messages from the subscribed channel(s) in real time
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"Received message: {message['data'].decode('utf-8')}")

def publisher(r, channel_name, message):
    """
    Publishes a message to the specified Redis channel.
    All subscribers to this channel immediately receive the message.
    """
    r.publish(channel_name, message)

# Example usage
channel = "updates"

# Start subscriber in a separate thread to avoid blocking the main thread
sub_thread = threading.Thread(target=subscriber, args=(r, channel))
sub_thread.start()

# Publish messages
publisher(r, channel, "Hello from Windows!")
publisher(r, channel, "Another update!")

pubsub = r.pubsub() creates a PubSub object. pubsub.subscribe(channel_name) instructs Redis to send messages from that channel to the PubSub object. pubsub.listen() is an infinite iterator, yielding messages as they come. r.publish(channel_name, message) sends a message to all subscribers of channel_name.

Expiring keys in Redis

Redis supports time-to-live (TTL). This is essential for temporary data like sessions or short-lived caches.

# EXPIRE: Set a 30-second expiration on a key
r.set("temp_key", "some value")
r.expire("temp_key", 30)

# SETEX: Combined set + expire in one command
r.setex("temp_key2", 60, "another value")

r.expire("temp_key", 30) sets a TTL of 30 seconds. After 30 seconds, temp_key is automatically deleted. r.setex("temp_key2", 60, "another value") is a shortcut that sets the key and expiration at the same time.

Example: Session expiration

Here’s an example of how you can use expiring keys to set a session to expire after a specified duration of time.

def store_session_with_expiry(user_id, token, ttl=3600):
    """
    Stores a session token for a specific user with a time-to-live (TTL).
    By default, the session expires after 1 hour (3600 seconds).
    """
    session_key = f"user:{user_id}:session"
    r.setex(session_key, ttl, token)

def get_session_with_expiry(user_id):
    """
    Retrieves the session token for the user. Returns None if the key doesn't exist
    or if it has expired.
    """
    session_key = f"user:{user_id}:session"
    token = r.get(session_key)
    return token.decode('utf-8') if token else None

# Usage
store_session_with_expiry(2001, "session_token_abc", 3600)
retrieved_token = get_session_with_expiry(2001)
print(f"Retrieved token: {retrieved_token}")

store_session_with_expiry uses setex under the hood to store user:{user_id}:session with a specified TTL. If you don’t pass a TTL, the key could remain indefinitely.

When the TTL expires, Redis removes the key automatically, making it impossible to retrieve the session.

Best Practices for Using Redis with Python

Handling Redis requires some considerations and best practices to ensure optimal performance and avoid common pitfalls.

Some tips to keep in mind include:

• Properly configure persistence options for data safety.
• Monitor memory usage to prevent out-of-memory errors.
• Use pipelining for efficient batch operations.
• Keep an eye on network latency as it can impact performance when working with a remote Redis server.
• Secure Redis connections with passwords and firewalls.

Conclusion

Redis is a powerful and versatile tool for managing data in Python applications. Using Python with Redis can bring about many applications and use cases, from caching data to managing queues. 

Keen to learn more about Redis? Our Introduction to NoSQL course can help you learn how to handle key-value data. Interested in Python instead? Then our Python Data Fundamentals skill track might be the best place to start.