Proyecto: publicar, consumir y reclamar

Continúa en la misma carpeta con almacen.py y el entorno virtual activo. Usaremos la instancia local de Redis del curso en el puerto 6380, inicialmente sin persistencia. Esto hace visible después que la intención registrada en SQLite permite reconstruir el transporte. Para operación real aplicarías también la configuración duradera del capítulo anterior.

export REDIS_URL=redis://127.0.0.1:6380/0
export INFORMES_DB=informes.db

Son variables de entorno leídas por los procesos Python. La URL indica protocolo, dirección, puerto y base lógica. No contiene credenciales en este laboratorio de loopback. Un despliegue con autenticación usaría secretos y evitaría imprimir esa URL.

La secuencia no tiene una transacción que abarque Redis y SQLite. En su lugar conserva información suficiente para repetir con seguridad: si SQLite registró la intención pero XADD falla, una reconciliación la encuentra; si el informe quedó guardado pero el ACK falla, la reentrega encuentra el mismo informe.

Los tres nombres compartidos son dirección, Stream y grupo. asegurar_grupo utiliza 0-0 para no saltarse entradas preexistentes y tolera únicamente BUSYGROUP, el error que significa que el grupo ya existe. Un error de permisos o conexión debe seguir siendo visible.

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

STREAM = "informes:trabajos"
GROUP = "generadores"


def conectar():
    return redis.Redis.from_url(
        os.environ.get("REDIS_URL", "redis://127.0.0.1:6380/0"),
        decode_responses=True, protocol=2,
        socket_connect_timeout=0.5, socket_timeout=2,
        retry=Retry(NoBackoff(), 0),
    )


def asegurar_grupo(r):
    try:
        r.xgroup_create(STREAM, GROUP, id="0-0", mkstream=True)
    except redis.ResponseError as error:
        if "BUSYGROUP" not in str(error):
            raise

El cliente se cerrará mediante with conectar() as r. Fijamos dos segundos de socket timeout, por encima del bloqueo de lectura de un segundo. Cero reintentos automáticos evita ocultar un XADD de resultado incierto; la recuperación la decide el protocolo de trabajo, con su job_id estable.

argparse pertenece a Python y transforma argumentos de terminal en valores con ayuda y validación. Con nargs="?", los dos argumentos posicionales son opcionales porque existe otro modo, --reconciliar; después comprobamos qué requiere cada modo. json.loads convierte la lista escrita en terminal y almacen.registrar valida el contrato.

El camino importante es:

payload = almacen.registrar(args.job_id, json.loads(args.muestras))
message_id = r.xadd(STREAM, {"job_id": args.job_id, "payload": payload})

Es un fragmento explicativo del archivo completo siguiente. Una caída entre ambas líneas deja una intención sin informe; no la borra. Una caída después de XADD puede dejar una entrada sin respuesta al productor. En ambos casos volver a publicar con la misma identidad es aceptable para nuestro receptor.

import argparse
import json
import almacen
from canal import STREAM, conectar, asegurar_grupo


def main():
    p = argparse.ArgumentParser()
    p.add_argument("job_id", nargs="?")
    p.add_argument("muestras", nargs="?")
    p.add_argument("--reconciliar", action="store_true")
    args = p.parse_args()
    if not args.reconciliar and (args.job_id is None or args.muestras is None):
        p.error("Indica job_id y una lista JSON de muestras")
    almacen.iniciar()
    if args.reconciliar:
        trabajos = almacen.pendientes()
    else:
        payload = almacen.registrar(args.job_id, json.loads(args.muestras))
        trabajos = [(args.job_id, payload)]
    with conectar() as r:
        asegurar_grupo(r)
        for job_id, payload in trabajos:
            message_id = r.xadd(STREAM, {"job_id": job_id, "payload": payload})
            print(job_id, message_id, flush=True)


if __name__ == "__main__":
    main()

--reconciliar publica hasta cien intenciones sin informe. Puede volver a publicar una que ya está siendo procesada: la deduplicación es deliberada. No es un barrido completo ni debe ejecutarse en un bucle apretado. Deja trabajar a los consumidores, mide lo que falta y vuelve a ejecutarlo con una cadencia controlada. Si algunas intenciones fallan siempre, investígalas en vez de inundar el Stream con copias.

El consumidor recibe un nombre, un modo y opciones de laboratorio. nuevos toma como máximo una entrada nunca entregada al grupo; recuperar reclama hasta diez pendientes elegibles. Ejecutar una tanda y terminar facilita observar cada transición. Para un servicio continuo envolverías ambas actividades en un ciclo con parada ordenada, espera, métricas y presupuesto de recuperación.

La decisión esencial dentro de procesar es:

estado, informe = almacen.aplicar(cuerpo["job_id"], cuerpo["payload"])
r.xack(STREAM, GROUP, message_id)

Si aplicar falla por SQLite ocupado, disco o conexión, el error sale y no se confirma. Un mensaje de esquema inválido sigue otro camino: primero almacen.rechazar confirma su evidencia y luego se hace ACK. Si guardar el rechazo falla, tampoco se retira de pendientes. Confirmar un rechazo no significa completar la intención original: es una condición operativa que queda visible.

import argparse
import json
import os
import almacen
from canal import STREAM, GROUP, conectar, asegurar_grupo


def procesar(r, message_id, cuerpo, fallo):
    if fallo == "antes":
        os._exit(21)
    try:
        if set(cuerpo) != {"job_id", "payload"} or len(cuerpo["payload"]) > 4096:
            raise ValueError("Mensaje inválido o demasiado grande")
        estado, informe = almacen.aplicar(cuerpo["job_id"], cuerpo["payload"])
    except (ValueError, TypeError, KeyError) as error:
        almacen.rechazar(message_id, cuerpo, error)
        r.xack(STREAM, GROUP, message_id)
        print("rechazado", message_id, str(error), flush=True)
        return
    print(estado, cuerpo["job_id"], json.dumps(informe), flush=True)
    if fallo == "despues":
        os._exit(22)
    r.xack(STREAM, GROUP, message_id)
    print("ack", message_id, flush=True)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("consumidor")
    p.add_argument("modo", choices=["nuevos", "recuperar"])
    p.add_argument("--fallo", choices=["no", "antes", "despues"], default="no")
    p.add_argument("--idle-ms", type=int, default=30000)
    p.add_argument("--cursor", default="0-0")
    args = p.parse_args()
    if args.idle_ms < 0:
        p.error("idle-ms no puede ser negativo")
    almacen.iniciar()
    with conectar() as r:
        asegurar_grupo(r)
        if args.modo == "nuevos":
            lotes = r.xreadgroup(GROUP, args.consumidor, {STREAM: ">"}, count=1, block=1000)
            mensajes = lotes[0][1] if lotes else []
        else:
            cursor, mensajes, borrados = r.xautoclaim(
                STREAM, GROUP, args.consumidor, args.idle_ms,
                start_id=args.cursor, count=10)
            print("siguiente_cursor", cursor, "perdidos_por_recorte", borrados, flush=True)
        for message_id, cuerpo in mensajes:
            procesar(r, message_id, cuerpo, args.fallo)
        if not mensajes:
            print("sin mensajes elegibles", flush=True)


if __name__ == "__main__":
    main()

Usa un identificador distinto del ensayo local del capítulo anterior:

python publicar.py lote-A '[10,20,30]'
python consumir.py ana nuevos
python consumir.py ana nuevos

El productor imprime el job_id y un ID Redis. La primera ejecución consumidora imprime creado lote-A, un resultado con media veinte y un ack. La segunda espera hasta un segundo y responde sin mensajes elegibles. Si publicas otra vez lote-A con el mismo contenido, habrá una entrada nueva pero el consumidor responderá repetido.

docker exec redis-practica redis-cli XPENDING informes:trabajos generadores
docker exec redis-practica redis-cli XLEN informes:trabajos

Los pendientes quedan en cero, mientras el Stream conserva las entradas históricas. No hemos aplicado TTL ni recortado el Stream. La siguiente página ensaya ambos intervalos de caída, colisiones de identidad, mensajes inválidos y pérdida completa del transporte antes de decidir cómo operarlo.