Python Redis: Guía para principiantes

Utilizar Redis con Python es una potente combinación para los ingenieros de datos que buscan construir sistemas de alto rendimiento, escalables y mantenibles. Tanto si estás optimizando capas de almacenamiento en caché, gestionando flujos de eventos en tiempo real o implementando el bloqueo distribuido, Redis proporciona una solución flexible y eficiente.

En esta guía, cubriremos los aspectos básicos del uso de Redis con Python para ayudarte a integrarlo perfectamente en tus flujos de trabajo de ingeniería de datos.

¿Qué es Python Redis?

Redis es un almacén de estructuras de datos en memoria de código abierto que se utiliza como base de datos, caché y agente de mensajes. Son las siglas de Servidor de Diccionario Remoto y se diseñó originalmente para ser un sistema de almacenamiento clave-valor con tiempos de acceso rápidos.

Admite varias estructuras de datos como cadenashashes, listas, conjuntos y conjuntos ordenados con consultas de rango. Redis también proporciona funciones avanzadas como transacciones, mensajería pub/sub, scripts Lua y capacidades integradas de replicación y agrupación para mejorar la disponibilidad y la escalabilidad.

Casos de uso habituales de Redis

Los posibles casos de uso de Redis son amplios y variados.

Algunos de los casos de uso más comunes en la ingeniería de datos son: 

• Almacenamiento en caché: Acelera las consultas a la base de datos y las respuestas a la API.
• Colas de mensajes: Potencia las arquitecturas basadas en eventos y las cadenas de procesamiento de trabajos.
• Análisis en tiempo real: Procesa y analiza datos en flujo con una latencia mínima.
• Tablas de clasificación: Clasifica y recupera eficazmente las entidades mejor valoradas.
• Almacenamiento de la sesión: Mantén una gestión del estado de la sesión rápida y escalable.
• Gestión de colas de trabajo: Garantizar una programación y ejecución eficaces de las tareas.
• Indexación geoespacial: Almacena y consulta datos basados en la ubicación.

Redis también se utiliza habitualmente en combinación con otras tecnologías como servidores web (por ejemplo, NGINX), bases de datos (por ejemplo, MySQL) y sistemas de mensajería (por ejemplo, Kafka) para optimizar el rendimiento y la escalabilidad.

¿Por qué usar Redis con Python?

Python es un lenguaje dominante en la ingeniería de datos debido a su flexibilidad, amplio ecosistema y facilidad de integración con las tecnologías de big data. 

Combinarlo con Redis puede aportar muchas ventajas a tus aplicaciones:

• Rendimiento rápido: Como Redis almacena los datos en memoria, puede ofrecer tiempos de acceso significativamente más rápidos en comparación con las bases de datos tradicionales bases de datos que almacenan los datos en disco. Esto lo hace ideal para casos de uso que requieren un alto rendimiento, como el almacenamiento en caché y las aplicaciones en tiempo real.
• Escalabilidad: Redis está diseñado pensando en la escalabilidad, lo que facilita el manejo de grandes cantidades de datos sin comprometer el rendimiento.
• Facilidad de uso: La sencilla sintaxis de Python y su rico ecosistema de bibliotecas facilitan el trabajo con Redis. La biblioteca cliente oficial Redis-py también proporciona una interfaz fácil de usar para interactuar con Redis desde tu código Python.

¿Qué es Redis-py?

Redis-py es la biblioteca cliente oficial de Redis para Python. Proporciona una interfaz fácil de usar para comunicarte con Redis desde tu código Python.

Algunas de sus principales características son:

• Soporte para todos los tipos de datos Redis
• Soporte de tuberías y transacciones para operaciones eficientes y atómicas
• Soporte de mensajería pub/sub
• Codificación/decodificación automática de datos a/desde bytes, cadenas y otros tipos de Python
• Compatibilidad con el modo clúster para trabajar con clústeres Redis

Configuración de Redis para el desarrollo en Python

Veamos cómo podemos configurar Redis nosotros mismos utilizando Python.

Hay tres formas principales de poner en marcha Redis:

1. Docker (Recomendado para ingenieros de datos)
2. Subsistema Windows para Linux (WSL)
3. Construcciones de terceros (por ejemplo Memurai, tporadowski/redis)

Método A: Utilizar Docker

En primer lugar, al utilizar Docker, necesitarás tener instalado Docker desktop en tu máquina. Se puede descargar desde sitio web de DockerHub.

Una vez instalado Docker, abre tu símbolo del sistema e introduce el siguiente comando.

docker pull redis

A continuación, iniciaremos un nuevo contenedor.

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

docker pull redis descarga la última imagen oficial de Redis. docker run --name my-redis -p 6379:6379 -d redis inicia un nuevo contenedor llamado mi-redis. El parámetro -p 6379:6379 asigna el puerto 6379 de tu máquina local al puerto 6379 del contenedor, para que puedas conectarte a Redis en localhost:6379. La bandera -d ejecuta el contenedor en segundo plano (modo separado).

Para verificarlo, ejecútalo:

docker exec -it my-redis redis-cli ping

Debería responder con PONG.

Método B: Utilizar el Subsistema Windows para Linux (WSL)

Para las máquinas Windows, otro método posible es utilizar el Subsistema Windows para la instalación. Esto utilizará Linux para realizar la instalación.

wsl --install

Este comando instala y configura la distribución por defecto de Ubuntu dentro de WSL. Reinicia si es necesario.

A continuación, abre tu terminal de Ubuntu (desde Microsoft Store o escribiendo wsl en PowerShell) y ejecútalo:

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

A continuación, inicia tu servidor Redis.

sudo service redis-server start

Hagamos una prueba para ver si el servidor está en funcionamiento.

redis-cli ping

Aquí tienes una descripción detallada de los comandos que hemos utilizado anteriormente:

• sudo apt-get update obtiene las últimas definiciones de paquetes.
• sudo apt-get install redis-server instala Redis en el entorno Ubuntu.
• sudo service redis-server start inicia inmediatamente el servicio Redis.
• redis-cli ping comprueba la conectividad con el servidor Redis local, devolviendo PONG si tiene éxito.

Método C: Construcciones de terceros

• Memurai: Memurai.com
• tporadowski/redis: Publicaciones en GitHub

Estas soluciones proporcionan binarios nativos de Windows. Tras descargarlo e instalarlo o descomprimirlo, ejecútalo:

redis-server.exe

Una vez iniciado, redis-server.exe debe imprimir registros que indiquen que está a la escucha en 127.0.0.1:6379.

Puedes confirmarlo utilizando la CLI incluida (si se proporciona):

redis-cli.exe ping

Debes esperar un resultado de PONG si tienes éxito.

Instalación de la biblioteca redis-py

A continuación, tendrás que instalar la biblioteca redis-py para poder acceder a la base de datos utilizando Python. Antes de instalar la biblioteca, asegúrate de que tienes Python y Redis instalados por separado.

pip install redis

Este comando descarga e instala la biblioteca cliente Redis de Python (redis-py) desde PyPI. Si utilizas un entorno virtual, asegúrate de activarlo antes de ejecutar este comando.

Verifícalo utilizando:

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

Esto debería imprimir un número de versión (por ejemplo, 5.2.1).

Configurar un servidor Redis

Si has seguido cualquiera de los métodos anteriores (Docker, WSL o una compilación de terceros), ahora deberías tener un servidor Redis ejecutándose en localhost:6379. Eso es suficiente para el desarrollo local en Windows.

Trabajar con estructuras de datos Redis

A continuación hay ejemplos en Python que demuestran cómo conectarse y manipular diferentes tipos de datos de Redis. Cada bloque de código incluye comentarios explicativos que detallan cómo funcionan los métodos o comandos.

Cadenas en 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")

Creamos un objeto Redis que se conecta al servidor Redis local.

• r.set("mykey", "hello from Windows") almacena la cadena "hola desde Windows" bajo la clave "miclave".
• r.get("mykey") recupera el valor almacenado en miclave. Redis devuelve bytes, así que lo descodificamos para obtener una salida legible por humanos.
• r.delete("mykey") borra completamente la clave miclave.

Ejemplo: Almacenar y recuperar sesiones de usuario

Aquí tienes un ejemplo de código de gestión de sesiones de usuario:

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): Crea un ID único global a través de uuid.uuid4() y lo almacena bajo una clave específica del usuario. get_user_session(user_id): Lee el testigo de sesión. Si None, la sesión no existe o ha caducado. delete_user_session(user_id): Elimina la clave de sesión de Redis, cerrando efectivamente la sesión del usuario.

Listas en Redis

Las Listas Redis son secuencias ordenadas de cadenas. Permiten operaciones en ambos extremos, lo que las hace útiles para colas, pilas o registros.

# 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") añade "task1" a la parte izquierda de una lista llamada "cola_tareas". rpush("task_queue", "task2") y rpush("task_queue", "task3") añaden tareas a la parte derecha. lpop("task_queue") and rpop("task_queue") elimina y devuelve elementos de los respectivos extremos de la lista.

Ejemplo: Implementar una cola simple respaldada por Redis

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 siempre añade nuevas tareas al final de la lista, lo que simula una cola FIFO tradicional. dequeue_task elimina tareas de la parte delantera. Si la lista está vacía, devuelve None.

El bucle while procesa las tareas hasta que la cola está vacía, y luego se detiene.

Hashes en Redis

Los hash son similares a los diccionarios de Python, pero almacenados en Redis. Son mejores para agrupar campos relacionados (por ejemplo, datos de usuario).

# 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") establece el campo "name" con el valor "Alicia" dentro del hash en la clave "usuario:1001". hget("user:1001", "email") recupera el campo "email". hdel("user:1001", "email") elimina el campo "email" de ese hash.

Ejemplo: Almacenamiento y acceso a perfiles de usuario estructurados

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)

Almacenamos el nombre y el correo electrónico de un usuario dentro de un hash. mapping={"name": name, "email": email} utiliza la capacidad de redis-pyde establecer varios campos a la vez. r.hgetall(user_key) recupera todos los campos y valores, devueltos como un diccionario de bytes en bruto que decodificamos a cadenas. delete_user_profile elimina toda la clave hash.

Conjuntos y conjuntos ordenados

Los datos en Redis pueden gestionarse mediante conjuntos o conjuntos ordenados. 

Los conjuntos son colecciones de valores únicos, mientras que los conjuntos ordenados son colecciones de pares clave-valor con una puntuación asociada a cada valor. Estas estructuras de datos proporcionan formas eficaces de almacenar y recuperar datos, lo que las hace útiles para aplicaciones que requieren búsquedas rápidas.

Establece

Un conjunto en Redis es similar a un conjunto matemático: contiene una colección de elementos únicos, y no se permiten elementos duplicados. Los conjuntos pueden utilizarse para modelar relaciones entre entidades diferentes, como los usuarios a los que les ha gustado una publicación en las redes sociales.

Aquí tienes una demostración de juegos:

# 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'}

Los Conjuntos Redis garantizan la unicidad de los miembros. Intentar añadir un miembro duplicado no tendrá ningún efecto.

El resultado de smembers es un conjunto de bytes Python.

Conjuntos ordenados

Los conjuntos ordenados en Redis son similares a los conjuntos normales, pero cada miembro tiene además una puntuación correspondiente. Esto permite realizar operaciones eficaces de ordenación y clasificación del conjunto.

Los conjuntos ordenados se utilizan a menudo para las tablas de clasificación, donde la puntuación representa el rango o los puntos de un jugador.

Aquí tienes una demostración de conjuntos ordenados:

# 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)]

Los conjuntos ordenados almacenan los miembros en un orden específico determinado por sus puntuaciones numéricas. zrange("leaderboard", 0, -1, withscores=True) devuelve todos los miembros desde el rango 0 hasta el final, incluyendo sus puntuaciones.

Ejemplo: Gestionar etiquetas o tablas de clasificación

El uso de conjuntos ordenados puede ser útil para gestionar etiquetas o crear tablas de clasificación. 

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 y get_tags muestran un sistema de etiquetado de uno a muchos, almacenando las etiquetas en un Conjunto. update_leaderboard y get_leaderboard muestran cómo podrías implementar una tabla de clasificación de un juego.

Funciones de Redis en aplicaciones Python

Redis tiene varias características que lo convierten en una opción popular para el almacenamiento de datos en aplicaciones Python: colas, bloqueo y almacenamiento en caché. 

A continuación te mostramos sus implementaciones en Python:

Redis como cola (con bloqueo)

Vamos a explorar el uso de Redis como cola con bloqueo utilizando BLPOP.

BLPOP es una operación de bloqueo que espera hasta que un elemento esté disponible en una lista.

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 utiliza blpop en un bucle. Si la lista está vacía, blpop esperará hasta que se introduzca otro elemento.

Una vez recibido un artículo, se elimina de la lista y se imprime.

Este enfoque es ideal para los modelos productor-consumidor, en los que los trabajadores consumen tareas a medida que aparecen.

Implementación de bloqueos en Redis

Los bloqueos distribuidos impiden que varios clientes o procesos modifiquen simultáneamente el mismo recurso. Redis puede ayudar a evitar condiciones de carrera en un entorno distribuido.

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()

Creamos un bloqueo con timeout=10, lo que significa que si el bloqueo no se libera manualmente, Redis lo eliminará automáticamente en 10 segundos para evitar el bloqueo indefinido (deadlock). lock.acquire(blocking=True, blocking_timeout=5) intenta adquirir el bloqueo durante 5 segundos antes de rendirse. Tras finalizar, lock.release() libera el recurso para otros procesos.

Almacenamiento en caché con Redis

El almacenamiento en caché es un escenario de uso común: almacenar en memoria los datos a los que se accede con frecuencia, reduciendo la carga de las bases de datos o las API externas.

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

Comprobamos si user_data:{user_id} existe en Redis. Si lo hace, se trata de una coincidencia con la caché, y nos saltamos la llamada a la API. Si no, lo obtenemos de la API remota y, a continuación, setex (conjunto + caducidad) durante 60 segundos. Las llamadas posteriores dentro de ese plazo recuperarán los datos almacenados en caché, reduciendo la latencia.

Temas avanzados en Redis

¿Crees que ya le has cogido el truco a Redis? Veamos algunos temas más avanzados.

Utilizar Pub/Sub con Redis

Pub/Sub (Publicar/Suscribir) es un popular patrón de mensajería para la comunicación en tiempo real. Los editores envían mensajes a un canal, y todos los suscriptores de ese canal reciben los mensajes.

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() crea un objeto PubSub. pubsub.subscribe(channel_name) ordena a Redis que envíe mensajes de ese canal al objeto PubSub. pubsub.listen() es un iterador infinito, que devuelve los mensajes a medida que llegan. r.publish(channel_name, message) envía un mensaje a todos los suscriptores de channel_name.

Caducidad de claves en Redis

Redis admite el tiempo de vida (TTL). Esto es esencial para datos temporales como sesiones o cachés de corta duración.

# 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) establece un TTL de 30 segundos. Transcurridos 30 segundos, temp_key se borra automáticamente. r.setex("temp_key2", 60, "another value") es un acceso directo que establece la clave y la caducidad al mismo tiempo.

Ejemplo: Vencimiento de la sesión

Aquí tienes un ejemplo de cómo puedes utilizar claves de caducidad para establecer que una sesión caduque tras un periodo de tiempo determinado.

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 utiliza setex bajo el capó para almacenar user:{user_id}:session con un TTL especificado. Si no pasas un TTL, la clave podría permanecer indefinidamente.

Cuando expira el TTL, Redis elimina la clave automáticamente, haciendo imposible recuperar la sesión.

Buenas prácticas para utilizar Redis con Python

Manejar Redis requiere algunas consideraciones y buenas prácticas para garantizar un rendimiento óptimo y evitar los errores más comunes.

Algunos consejos a tener en cuenta son

• Configura correctamente las opciones de persistencia para la seguridad de los datos.
• Controla el uso de la memoria para evitar errores por falta de memoria.
• Utiliza el pipelining para operaciones por lotes eficientes.
• Vigila la latencia de la red, ya que puede afectar al rendimiento cuando trabajes con un servidor Redis remoto.
• Protege las conexiones Redis con contraseñas y cortafuegos.

Conclusión

Redis es una herramienta potente y versátil para gestionar datos en aplicaciones Python. Usar Python con Redis puede dar lugar a muchas aplicaciones y casos de uso, desde almacenar datos en caché hasta gestionar colas. 

¿Quieres saber más sobre Redis? Nuestra Introducción a NoSQL puede ayudarte a aprender a manejar datos clave-valor. ¿Te interesa Python? Entonces nuestro Fundamentos de Datos en Python puede ser el mejor lugar para empezar.