45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
import json
|
|
import os
|
|
import redis
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CHANNEL_PREFIX = "audit_task:"
|
|
STATE_PREFIX = "audit_task_state:"
|
|
STATE_TTL = 7200 # 2 horas
|
|
|
|
|
|
def _get_sync_redis():
|
|
# REDIS_PUBSUB_HOST apunta al Redis compartido con el backend Django.
|
|
# En dev: redis_backend_dev. Por defecto usa REDIS_HOST (para entornos con una sola instancia).
|
|
return redis.Redis(
|
|
host=os.getenv("REDIS_PUBSUB_HOST", os.getenv("REDIS_HOST", "localhost")),
|
|
port=int(os.getenv("REDIS_PUBSUB_PORT", os.getenv("REDIS_PORT", 6379))),
|
|
db=int(os.getenv("REDIS_PUBSUB_DB", os.getenv("REDIS_DB", 0))),
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
socket_timeout=2,
|
|
)
|
|
|
|
|
|
def publish_task_event(task_id: str, status: str, message: str = "", resultado: dict = None, progress: int = None):
|
|
"""
|
|
Publica un evento de progreso de tarea en Redis Pub/Sub.
|
|
Guarda el último estado en una key con TTL para clientes que se conectan tarde.
|
|
"""
|
|
payload: dict = {"task_id": task_id, "status": status, "message": message}
|
|
if resultado is not None:
|
|
payload["resultado"] = resultado
|
|
if progress is not None:
|
|
payload["progress"] = progress
|
|
|
|
try:
|
|
client = _get_sync_redis()
|
|
serialized = json.dumps(payload)
|
|
client.publish(f"{CHANNEL_PREFIX}{task_id}", serialized)
|
|
client.setex(f"{STATE_PREFIX}{task_id}", STATE_TTL, serialized)
|
|
client.close()
|
|
except Exception as exc:
|
|
logger.error(f"[redis_events] Error publicando evento para tarea {task_id}: {exc}")
|