hermes
131 lines
4.9 KiB
Python
131 lines
4.9 KiB
Python
"""Queue-Handler für Redis-basierte Message-Queue mit Idempotenz-Garantie."""
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import redis
|
|
|
|
from src.models.submission import Submission
|
|
|
|
|
|
class QueueHandler:
|
|
"""Redis-basierte Message-Queue mit Idempotenz-Garantie.
|
|
|
|
Verwendet Redis-Streams für persistente Pufferung und ein Redis-Set
|
|
für Deduplizierung mit 24h TTL.
|
|
"""
|
|
|
|
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379) -> None:
|
|
"""Initialisiert den Queue-Handler mit Redis-Verbindung.
|
|
|
|
Args:
|
|
redis_host: Redis-Server-Host (Standard: localhost)
|
|
redis_port: Redis-Server-Port (Standard: 6379)
|
|
"""
|
|
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
|
|
self.stream_key = "bsn:submissions"
|
|
self.processed_key = "bsn:processed"
|
|
self.consumer_group = "bsn-workers"
|
|
|
|
# Consumer-Group initialisieren (falls noch nicht vorhanden)
|
|
try:
|
|
self.redis.xgroup_create(self.stream_key, self.consumer_group, id="0")
|
|
except redis.exceptions.ResponseError:
|
|
pass # Group existiert bereits
|
|
|
|
def enqueue(self, submission: Submission) -> str | None:
|
|
"""Legt eine Submission in die Queue. Überspringt Duplikate.
|
|
|
|
Prüft ob message_id bereits verarbeitet wurde.
|
|
Wenn ja → return None (Idempotenz)
|
|
Wenn nein → XADD in Stream, merke ID in processed_set (TTL 24h)
|
|
|
|
Args:
|
|
submission: Die zu enqueuende Submission.
|
|
|
|
Returns:
|
|
Die Redis-Stream-ID bei Erfolg, None bei Duplikat.
|
|
"""
|
|
# Prüfe ob message_id bereits verarbeitet wurde
|
|
if self.redis.sismember(self.processed_key, submission.message_id):
|
|
return None
|
|
|
|
# Submission als dict serialisieren
|
|
data: dict[str, Any] = {
|
|
"message_id": submission.message_id,
|
|
"plattform": submission.plattform,
|
|
"absender": submission.absender,
|
|
"text": submission.text,
|
|
"zeitstempel": submission.zeitstempel.isoformat(),
|
|
"status": submission.status,
|
|
}
|
|
|
|
# XADD in Stream
|
|
stream_id = self.redis.xadd(self.stream_key, data)
|
|
|
|
# In processed_set merken mit 24h TTL (86400 Sekunden)
|
|
self.redis.sadd(self.processed_key, submission.message_id)
|
|
self.redis.expire(self.processed_key, 86400)
|
|
|
|
# stream_id kann bytes oder str sein, aber mit decode_responses=True ist es str
|
|
return str(stream_id) if stream_id else None
|
|
|
|
def dequeue(self, count: int = 5) -> list[Submission]:
|
|
"""Holt bis zu `count` Submissions aus der Queue (FIFO).
|
|
|
|
Nutzt XREADGROUP für Consumer-Group-Verarbeitung.
|
|
|
|
Args:
|
|
count: Maximale Anzahl zu holender Submissions (Standard: 5).
|
|
|
|
Returns:
|
|
Liste von Submission-Objekten.
|
|
"""
|
|
messages: list[Any] = self.redis.xreadgroup(
|
|
groupname=self.consumer_group,
|
|
consumername="worker-1", # Einfacher Einzel-Consumer
|
|
streams={self.stream_key: ">"}, # ">" = nur neue Nachrichten
|
|
count=count,
|
|
block=0, # Nicht blockierend
|
|
)
|
|
|
|
submissions: list[Submission] = []
|
|
if messages:
|
|
for stream_name, stream_messages in messages:
|
|
for message_tuple in stream_messages:
|
|
# message_tuple ist (id, {field: value, ...})
|
|
message_data = message_tuple[1]
|
|
try:
|
|
zeitstempel_str = str(message_data.get("zeitstempel", ""))
|
|
submission = Submission(
|
|
message_id=str(message_data.get("message_id", "")),
|
|
plattform=str(message_data.get("plattform", "")),
|
|
absender=str(message_data.get("absender", "")),
|
|
text=str(message_data.get("text", "")),
|
|
zeitstempel=datetime.fromisoformat(zeitstempel_str),
|
|
status=str(message_data.get("status", "neu")),
|
|
)
|
|
submissions.append(submission)
|
|
except (KeyError, ValueError, TypeError):
|
|
continue # Ungültige Nachricht überspringen
|
|
|
|
return submissions
|
|
|
|
def mark_done(self, submission: Submission) -> None:
|
|
"""Markiert eine Submission als erfolgreich verarbeitet.
|
|
|
|
Entfernt die Nachricht aus dem Stream-Aktivitäts-Tracking.
|
|
|
|
Args:
|
|
submission: Die abzuschließende Submission.
|
|
"""
|
|
# Nachricht im Stream ack-en (für Consumer-Group)
|
|
self.redis.xack(self.stream_key, self.consumer_group, submission.message_id)
|
|
|
|
def queue_size(self) -> int:
|
|
"""Gibt die aktuelle Queue-Größe zurück.
|
|
|
|
Returns:
|
|
Anzahl wartender Nachrichten im Stream.
|
|
"""
|
|
return self.redis.xlen(self.stream_key)
|