Conectar Python sin esconder los fallos de red

redis-py es la biblioteca de Python que envía comandos al servidor. Se instala con el nombre de paquete redis; el import también se llama redis. No instala ni arranca redis-server. Continúa con la instancia del capítulo inicial en 127.0.0.1:6380.

python3 -m venv .venv
. .venv/bin/activate
python -m pip install redis==8.1.0

Un entorno virtual guarda los paquetes de esta práctica aparte de los del sistema. Actívalo en cada terminal que ejecute los ejemplos. Guarda el siguiente bloque como conexion.py y ejecútalo con python conexion.py:

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry

r = redis.Redis(
    host="127.0.0.1", port=6380, db=0,
    decode_responses=True, protocol=2,
    socket_connect_timeout=0.5,
    socket_timeout=1.0,
    max_connections=10,
    retry=Retry(NoBackoff(), 0),
)
print(r.ping())
r.set("curso:python", "hola", ex=60)
print(r.get("curso:python"))
r.close()

Espera True y hola. decode_responses=True convierte respuestas de bytes a texto UTF-8. protocol=2 fija una forma conocida de las respuestas de los ejemplos; Redis y el cliente también soportan RESP3. Elegir uno no cambia las garantías de persistencia. db=0 selecciona una base lógica, no otro proceso ni otro presupuesto de memoria.

Retry representa la política de reintentos y NoBackoff una espera nula entre ellos. Aquí fijamos cero reintentos automáticos para que un timeout no repita silenciosamente una escritura. Más adelante decidirás qué operaciones permiten reintento y con qué identidad.

import json
import redis

r = redis.Redis(host="127.0.0.1", port=6380, decode_responses=True)
producto = {"id": 42, "nombre": "Lámpara", "activo": True}
r.set("curso:json", json.dumps(producto, ensure_ascii=False), ex=60)
raw = r.get("curso:json")
recuperado = json.loads(raw) if raw is not None else None
print(recuperado["nombre"])
print(r.get("curso:no-existe") is None)
r.close()

json pertenece a la biblioteca estándar: dumps convierte valores Python en una cadena JSON y loads realiza el camino contrario. ensure_ascii=False mantiene los caracteres legibles. JSON no serializa automáticamente fechas, decimales ni cualquier objeto Python. Decide un contrato sencillo, por ejemplo fechas ISO como texto e importes enteros.

No confundas None con "null", "" o "0". Solo raw is None identifica la ausencia. Una caché de “no encontrado” necesita un valor que represente esa situación sin confundirse con un miss. No uses pickle para datos que otro actor pueda manipular: deserializarlo puede ejecutar código. Para estos ejemplos bastan datos JSON validados.

El cliente mantiene un pool, una reserva de conexiones reutilizables. Una operación toma una, envía el comando, recibe la respuesta y la devuelve. Crea el cliente una vez por proceso de aplicación y compártelo donde corresponda, en lugar de construir y cerrar uno por petición HTTP.

max_connections=10 pone un techo; con el pool normal, agotarlo produce un error en lugar de esperar indefinidamente. Un BlockingConnectionPool permite esperar un tiempo limitado cuando eso convenga. No necesitas cien conexiones porque atiendas cien usuarios: importa cuántas operaciones están usando Redis a la vez. Un comando bloqueante como XREADGROUP BLOCK ocupa una conexión mientras espera, así que los trabajadores merecen su propio cliente/pool.

Los clientes síncronos pueden compartirse entre hilos según el contrato de la biblioteca, pero los objetos Pipeline y PubSub no deben compartirse de esa manera. En procesos creados por un servidor de aplicaciones, inicializa los recursos conforme al ciclo de vida de cada proceso; no dependas de sockets heredados accidentalmente.

Límite	Pregunta que contesta	Lo que no garantiza
Conexión	¿Cuánto espero para establecer el canal?	Un plazo para todo el trabajo
Socket	¿Cuánto espero en una operación de red?	Que una escritura no llegara a ejecutarse
Presupuesto de aplicación	¿Cuánto tiempo queda para responder al usuario?	Cancelar efectos ya enviados

Si SET sale del cliente, Redis lo aplica y se corta la respuesta, el cliente ve un timeout aunque la escritura exista. Reintentar SET con el mismo valor puede ser aceptable; reintentar INCR puede contar dos veces. La recuperación debe partir del significado del comando y del contrato de idempotencia, no de “falló, repito todo”.

try:
    valor = r.get("tienda:producto:v2:42")
except (redis.ConnectionError, redis.TimeoutError):
    valor = None  # Solo válido si es una caché reconstruible.

Este fragmento presupone un cliente r abierto y deja pendiente cómo consultar el origen con límites; lo resolveremos en la caché. No captures Exception para transformar un permiso denegado, un JSON corrupto o un error de programación en un miss silencioso. Esos fallos exigen diagnóstico.

redis.asyncio ofrece métodos que se esperan con await; no convierte Redis en otra base ni vuelve concurrentes tus reglas de negocio. Un cliente síncrono dentro de una corrutina puede bloquear su bucle de eventos. Si todavía no conoces corrutinas, sigue con la versión síncrona y consulta el curso de asincronía cuando necesites varias esperas simultáneas.

import asyncio
import redis.asyncio as redis_async

async def main():
    r = redis_async.Redis(host="127.0.0.1", port=6380, decode_responses=True)
    try:
        print(await r.ping())
    finally:
        await r.aclose()

asyncio.run(main())

finally cierra el cliente incluso si falla el PING. Una aplicación web duradera lo crearía al arrancar y lo cerraría al parar, no en cada petición. Mantén las mismas decisiones de límites, permisos e idempotencia que en la versión síncrona.

La guía oficial de redis-py y su referencia de conexiones documentan el pool, los parámetros y los clientes. Comprueba el fallo apagando solo tu contenedor de práctica, observando la excepción y volviéndolo a arrancar: el siguiente capítulo decide qué hacer con ella.