Redis es una base de datos no relacional (NoSQL) que almacena datos en memoria, lo que le confiere una velocidad de acceso y escritura significativamente superior a las bases de datos relacionales tradicionales que guardan información en disco. Su arquitectura permite manejar hasta 300,000 operaciones de lectura/escritura por segundo.

Operaciones con Tipos de Datos String

A continuación, se detallan las operaciones básicas sobre el tipo de dato String en Redis:

import redis

# Conexión a una instancia de Redis
# Se recomienda reemplazar la IP, contraseña y puerto por los de tu configuración
r = redis.Redis(host='localhost', port=6379, password='your_password', db=8)

# Añadir o actualizar un valor asociado a una clave
r.set('user:1', 'Alice')

# Añadir un valor con tiempo de expiración (en segundos)
r.setex('session:abc', 3600, 'session_data') # La clave expira en 1 hora

# Obtener el valor de una clave
user_data_binary = r.get('user:1')
if user_data_binary:
    user_data_string = user_data_binary.decode('utf-8') # Decodificar de bytes a string
    print(f"Valor de 'user:1': {user_data_string}")

# Eliminar una clave
r.delete('user:1')

# Obtener todas las claves que coincidan con un patrón
print(f"Todas las claves: {r.keys()}")
print(f"Claves que empiezan con 'user:': {r.keys('user:*')}")
print(f"Claves que contienen 'session': {r.keys('*session*')}")

# Intentar obtener una clave que no existe devuelve None
non_existent_data = r.get('non_existent_key')
print(f"Valor de clave inexistente: {non_existent_data}")

# Las claves pueden simular estructuras de carpetas usando prefijos con ':'
r.set('user:profile:1', 'Details')
print(f"Valor de 'user:profile:1': {r.get('user:profile:1').decode('utf-8')}")

# Eliminar todas las claves (¡usar con precaución!)
# for key in r.keys():
#     r.delete(key)

Operaciones con Tipos de Datos Hash

Los hashes en Redis son ideales para almacenar objetos compuestos.

# Añadir campos y valores a un hash
r.hset('product:101', 'name', 'Laptop')
r.hset('product:101', 'price', '1200.50')
r.hset('product:101', 'stock', '50')

# Obtener el tipo de dato de una clave
print(f"Tipo de 'product:101': {r.type('product:101').decode('utf-8')}")

# Obtener un campo específico de un hash
product_name_binary = r.hget('product:101', 'name')
if product_name_binary:
    print(f"Nombre del producto: {product_name_binary.decode('utf-8')}")

# Obtener todos los campos y valores de un hash
all_product_details_binary = r.hgetall('product:101')
all_product_details = {}
if all_product_details_binary:
    for key, value in all_product_details_binary.items():
        all_product_details[key.decode('utf-8')] = value.decode('utf-8')
    print(f"Detalles del producto: {all_product_details}")

# Eliminar un campo específico de un hash
r.hdel('product:101', 'stock')

# Eliminar un hash completo
r.delete('product:101')

# Configurar tiempo de expiración para una clave (aplicable a todos los tipos)
r.expire('some_key', 600) # Expira en 10 minutos
print(f"Tiempo restante para 'some_key': {r.ttl('some_key')} segundos")

Conversión de Bytes a Strings

Redis devuelve los valores como objetos bytes. Es necesario decodificarlos para trabajar con ellos como strings legibles.

string_data = "Ejemplo de texto"
bytes_data = string_data.encode('utf-8') # Convertir string a bytes
print(f"Bytes: {bytes_data}")

bytes_to_decode = b"Datos en bytes"
decoded_string = bytes_to_decode.decode('utf-8') # Convertir bytes a string
print(f"String decodificado: {decoded_string}")

Entegración con Bases de Datos Relacionales

Ejemplo de cómo obtener datos de una base de datos MySQL y almacenarlos en Redis como hashes, serializando cada registro como JSON.

import pymysql
import json
import redis

# Configuración de conexión a Redis
redis_client = redis.Redis(host='localhost', port=6379, password='your_password', db=1)

# Configuración de conexión a MySQL
db_connection = pymysql.connect(
    host='mysql_host',
    user='mysql_user',
    password='mysql_password',
    database='your_database',
    charset='utf8'
)
cursor = db_connection.cursor(cursor=pymysql.cursors.DictCursor)

# Consulta SQL
cursor.execute('SELECT id, username, passwd, is_admin FROM users;')
all_users_data = cursor.fetchall()

# Procesar y almacenar datos en Redis
for user_record in all_users_data:
    username_key = user_record.get('username')
    if username_key:
        # Serializar el diccionario completo a JSON y almacenarlo como valor string
        redis_client.hset('user_profiles', username_key, json.dumps(user_record))

# Cerrar conexiones
cursor.close()
db_connection.close()

Clase de Abstracción para Operaciones Redis

Se presanta una clase que encapsula operaciones comunes de Redis para facilitar su uso.

import redis

class RedisManager:
    def __init__(self, host, password, port=6379, db=8):
        try:
            self.redis_conn = redis.Redis(host=host, port=port, password=password, db=db)
            # Verificar la conexión
            self.redis_conn.ping()
            print("Conexión a Redis establecida exitosamente.")
        except redis.exceptions.ConnectionError as e:
            print(f"Error al conectar con Redis: {e}")
            self.redis_conn = None

    def get_string(self, key):
        if not self.redis_conn: return None
        value = self.redis_conn.get(key)
        return value.decode('utf-8') if value else None

    def set_string(self, key, value, expiry_seconds=None):
        if not self.redis_conn: return False
        try:
            self.redis_conn.set(key, value, ex=expiry_seconds)
            return True
        except Exception as e:
            print(f"Error al guardar string en Redis: {e}")
            return False

    def delete_key(self, key):
        if not self.redis_conn: return False
        if self.redis_conn.exists(key):
            self.redis_conn.delete(key)
            print(f"Clave '{key}' eliminada.")
            return True
        else:
            print(f"La clave '{key}' no existe.")
            return False

    def get_hash_field(self, hash_name, field):
        if not self.redis_conn: return None
        value = self.redis_conn.hget(hash_name, field)
        return value.decode('utf-8') if value else None

    def set_hash_field(self, hash_name, field, value):
        if not self.redis_conn: return False
        try:
            self.redis_conn.hset(hash_name, field, value)
            return True
        except Exception as e:
            print(f"Error al guardar campo hash en Redis: {e}")
            return False

    def get_all_hash_fields(self, hash_name):
        if not self.redis_conn: return {}
        data_bytes = self.redis_conn.hgetall(hash_name)
        decoded_data = {}
        if data_bytes:
            for k, v in data_bytes.items():
                decoded_data[k.decode('utf-8')] = v.decode('utf-8')
        return decoded_data

    def delete_hash_field(self, hash_name, field):
        if not self.redis_conn: return False
        deleted_count = self.redis_conn.hdel(hash_name, field)
        if deleted_count:
            print(f"Campo '{field}' eliminado del hash '{hash_name}'.")
            return True
        else:
            print(f"No se pudo eliminar el campo '{field}' del hash '{hash_name}'. Podría no existir.")
            return False

    def clear_database(self):
        if not self.redis_conn: return False
        try:
            self.redis_conn.flushdb()
            print("Base de datos de Redis vaciada exitosamente.")
            return True
        except Exception as e:
            print(f"Error al vaciar la base de datos de Redis: {e}")
            return False

# Ejemplo de uso de la clase
# Reemplaza con tus credenciales de Redis
redis_manager = RedisManager(host='localhost', port=6379, password='your_password', db=8)

# Operaciones de ejemplo
redis_manager.set_string("app:config", "setting_value", expiry_seconds=300)
config = redis_manager.get_string("app:config")
print(f"Configuración: {config}")

redis_manager.set_hash_field("user:123", "name", "Bob")
redis_manager.set_hash_field("user:123", "email", "bob@example.com")
user_info = redis_manager.get_all_hash_fields("user:123")
print(f"Información de usuario: {user_info}")

# redis_manager.clear_database() # Descomentar para vaciar la BD