Redis CRUD Operations in Python¶

Redis CRUD Operations in Python involve using the redis-py library to interact with a Redis data store, commonly integrated with Python web frameworks like [[Flask]] to manage application data^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

This guide draws from a "Customer App" example that transitions from file-based storage (using json and csv) to an in-memory database structure^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

Prerequisites and Setup¶

To perform CRUD operations, the redis-py client is required^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md]. In a production environment, it is common to use Redis Sentinel for high availability, which handles automatic failover between master and slave nodes^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

Connection details, such as sentinel addresses, master name, and passwords, are typically injected via environment variables^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

from redis.sentinel import Sentinel
import os

# Configuration via Environment Variables
redis_sentinels = os.environ.get('REDIS_SENTINELS')
redis_master_name = os.environ.get('REDIS_MASTER_NAME')
redis_password = os.environ.get('REDIS_PASSWORD')

# Parse Sentinel List
sentinels = []
for s in redis_sentinels.split(","):
    sentinels.append((s.split(":")[0], s.split(":")[1]))

# Initialize Sentinel and Master Client
redis_sentinel = Sentinel(sentinels, socket_timeout=5)
redis_master = redis_sentinel.master_for(
    redis_master_name,
    password=redis_password,
    socket_timeout=5
)

Retry Logic¶

Because Redis nodes can fail over (e.g., the master node changes), a retry wrapper is recommended to handle temporary connection errors or timeouts^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

import redis.exceptions
import time

def redis_command(command, *args):
    max_retries = 3
    count = 0
    backoffSeconds = 5
    while True:
        try:
            return command(*args)
        except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError):
            count += 1
            if count > max_retries:
                raise
        print('Retrying in {} seconds'.format(backoffSeconds))
        time.sleep(backoffSeconds)

Create and Update¶

Data in Redis is often stored as stringified JSON^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md]. The Create (and Update) operation is handled by the set command.

import json

class Customer:
    def __init__(self, c="", f="", l=""):
        self.customerID = c
        self.firstName = f
        self.lastName = l
    # ... other methods ...

def updateCustomer(customer):
    # Serialize object to JSON and store in Redis
    redis_command(
        redis_master.set,
        customer.customerID,
        json.dumps(customer.__dict__)
    )

Read¶

Read operations can be performed in two ways: retrieving a specific item by key or scanning for all keys^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

Read by ID¶

To retrieve a single record, use the get command. Since Redis stores strings and bytes, the result must be decoded and deserialized from JSON^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

def getCustomer(customerID):
    customer = redis_command(redis_master.get, customerID)

    if customer == None:
        return {}
    else:
        # Decode bytes to string and parse JSON
        c = str(customer, 'utf-8')
        return json.loads(c)

Read All¶

To list all data (e.g., all customers), iterate over keys found using scan_iter^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

def getCustomers():
    customers = {}
    # Scan for keys matching a pattern
    customerIDs = redis_command(redis_master.scan_iter, "*")

    for customerID in customerIDs:
        # Retrieve data for each key
        customer = getCustomer(customerID)
        customers[customer["customerID"]] = customer

    return customers

Delete¶

While the provided source focuses on Create, Read, and Update, Delete operations in Redis are typically performed using the del command (e.g., redis_master.delete(key)). The provided source includes a cleanup step using FLUSHALL to remove test records, but does not implement a specific function for deleting individual customer records in the application logic^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md].

Sources¶

^[400-devops__09-Scripting-Language__python__introduction__part-5.database.redis__README.md]