Redis Caching in Python - Rishan Solutions

Redis (Remote Dictionary Server) is an in-memory key-value store that can be used as a database, cache, and message broker. It provides high-speed data access, making it an excellent choice for caching in Python applications.

1. Why Use Redis for Caching?

Lightning-fast – Data is stored in RAM, enabling quick retrieval.
Persistence – Optionally stores data on disk.
Supports Multiple Data Structures – Strings, Hashes, Lists, Sets, Sorted Sets.
Scalability – Can handle millions of requests per second.
Expiration & TTL – Automatically expires cached data.

2. Installing Redis and Required Python Packages

Install Redis on Your System

For Windows:

Download Redis from Memurai or use WSL to install Redis.

For Linux/macOS:

sudo apt update && sudo apt install redis-server -y  # Ubuntu/Debian
brew install redis  # macOS

Start Redis:

redis-server

Install Redis Python Client

pip install redis

3. Connecting to Redis in Python

import redis

# Connect to Redis Server
redis_client = redis.Redis(host='localhost', port=6379, db=0)

print("Connected to Redis!")

host='localhost' → Connects to local Redis instance.
port=6379 → Default Redis port.
db=0 → Default database (Redis supports multiple databases).

4. Storing and Retrieving Data in Redis

Set and Get a Key-Value Pair

redis_client.set("username", "narendra")
print(redis_client.get("username").decode())  # Output: narendra

.set(key, value) → Stores data in Redis.
.get(key) → Retrieves data from Redis.

5. Setting Expiry for Cached Data

Set a Key with Expiry (TTL – Time to Live)

redis_client.setex("session_token", 60, "xyz123")  # Expires in 60 seconds
print(redis_client.ttl("session_token"))  # Check time left before expiry

.setex(key, seconds, value) → Stores a key with an expiration time.
.ttl(key) → Returns remaining time before expiry.

6. Using Redis as a Cache for Expensive Operations

Without Redis Caching (Slow Execution)

import time

def slow_function():
    time.sleep(5)  # Simulate slow processing
    return "Processed Data"

start_time = time.time()
result = slow_function()
print("Result:", result, "Time Taken:", time.time() - start_time)

Each call takes 5 seconds!

With Redis Caching (Fast Execution)

def cached_function():
    cache_key = "expensive_operation"

    # Check if data exists in cache
    if redis_client.exists(cache_key):
        print("Fetching from Cache...")
        return redis_client.get(cache_key).decode()

    print("Computing Result...")
    result = slow_function()  # Expensive function
    redis_client.setex(cache_key, 300, result)  # Cache for 5 minutes
    return result

start_time = time.time()
result = cached_function()
print("Result:", result, "Time Taken:", time.time() - start_time)

First call is slow, but subsequent calls are instant!

7. Storing Complex Data Structures in Redis

Storing and Retrieving Python Dictionary (JSON)

import json

user_data = {"name": "Narendra", "age": 25, "city": "Hyderabad"}

# Convert to JSON and store
redis_client.set("user:1", json.dumps(user_data))

# Retrieve and convert back to dictionary
retrieved_data = json.loads(redis_client.get("user:1"))
print(retrieved_data)

Redis stores data as strings, so we use json.dumps() and json.loads().

8. Using Redis Hashes for Structured Data

Storing Data Using Redis Hashes

redis_client.hset("user:1001", mapping={"name": "John", "email": "john@example.com"})
print(redis_client.hgetall("user:1001"))  # Get all fields

.hset() stores multiple fields in a Redis hash.
.hgetall() retrieves all fields from a hash.

9. Implementing a Least Recently Used (LRU) Cache with Redis

Redis supports LRU eviction to remove old data when memory is full.

Configuring LRU Cache in Redis

redis-cli
CONFIG SET maxmemory 100mb
CONFIG SET maxmemory-policy allkeys-lru

This sets a 100MB cache limit and removes least recently used keys first.

10. Using Redis for Pub/Sub Messaging

Publisher (Send Message)

redis_client.publish("updates", "New blog post available!")

Publishes a message to the “updates” channel.

Subscriber (Receive Message)

pubsub = redis_client.pubsub()
pubsub.subscribe("updates")

for message in pubsub.listen():
    if message["type"] == "message":
        print("Received:", message["data"].decode())

Listens for new messages and prints them.

11. Deleting Cache Data

redis_client.delete("username")  # Delete a specific key
redis_client.flushdb()  # Clear all keys from the current Redis database

Use flushdb() cautiously, as it deletes all keys in the database!