some basic restructure

Author: Tanmaypatil123

modelq/app/backends/base.py

@@ -0,0 +1,119 @@
+from abc import ABC, abstractmethod
+from typing import Optional, Dict, Any
+
+
+class QueueBackend(ABC):
+    """Abstract interface for ModelQ queue backends."""
+
+    # ─── Task Enqueue/Dequeue ──────────────────────────────────────────────
+
+    @abstractmethod
+    def enqueue_task(self, task_data: dict) -> None:
+        """Push a new task to the task queue."""
+        pass
+
+    @abstractmethod
+    def dequeue_task(self, timeout: Optional[int] = None) -> Optional[dict]:
+        """Pop the next task from the queue (blocking or timed)."""
+        pass
+
+    @abstractmethod
+    def requeue_task(self, task_data: dict) -> None:
+        """Re-queue an existing task (e.g., after failure or rejection)."""
+        pass
+
+    @abstractmethod
+    def enqueue_delayed_task(self, task_data: dict, delay_seconds: int) -> None:
+        """Push task to delayed queue (sorted by run timestamp)."""
+        pass
+
+    @abstractmethod
+    def dequeue_ready_delayed_tasks(self) -> list:
+        """Get all delayed tasks ready to run now (score <= time.time())."""
+        pass
+
+    @abstractmethod
+    def flush_queue(self) -> None:
+        """Empty all tasks from the main task queue (for tests/dev reset)."""
+        pass
+
+    # ─── Task Status Management ────────────────────────────────────────────
+
+    @abstractmethod
+    def save_task_state(self, task_id: str, task_data: dict, result: bool) -> None:
+        """Save or update the final state of a task (completed/failed/etc)."""
+        pass
+
+    @abstractmethod
+    def load_task_state(self, task_id: str) -> Optional[dict]:
+        """Fetch a task's full state from storage."""
+        pass
+
+    @abstractmethod
+    def remove_task_from_queue(self, task_id: str) -> bool:
+        """Remove task from queue if still queued."""
+        pass
+
+    @abstractmethod
+    def mark_processing(self, task_id: str) -> bool:
+        """Add task to 'processing' set; return False if already processing."""
+        pass
+
+    @abstractmethod
+    def unmark_processing(self, task_id: str) -> None:
+        """Remove task from processing set."""
+        pass
+
+    @abstractmethod
+    def get_all_processing_tasks(self) -> list:
+        """Return list of currently 'processing' task IDs."""
+        pass
+
+    @abstractmethod
+    def get_all_queued_tasks(self) -> list:
+        """Return list of all tasks in the main queue."""
+        pass
+
+    # ─── Server Registry ───────────────────────────────────────────────────
+
+    @abstractmethod
+    def register_server(self, server_id: str, task_names: list) -> None:
+        """Register a worker with allowed task names and heartbeat."""
+        pass
+
+    @abstractmethod
+    def update_server_status(self, server_id: str, status: str) -> None:
+        """Update current server status and heartbeat time."""
+        pass
+
+    @abstractmethod
+    def get_all_server_ids(self) -> list:
+        """Return all currently registered server IDs."""
+        pass
+
+    @abstractmethod
+    def get_server_data(self, server_id: str) -> Optional[dict]:
+        """Get full data object for a server."""
+        pass
+
+    @abstractmethod
+    def prune_dead_servers(self, timeout: int) -> list:
+        """Remove any servers whose heartbeat is older than `timeout` seconds."""
+        pass
+
+    # ─── Metrics + Maintenance ─────────────────────────────────────────────
+
+    @abstractmethod
+    def prune_old_results(self, older_than_seconds: int) -> int:
+        """Delete old task results beyond TTL."""
+        pass
+
+    @abstractmethod
+    def queue_length(self) -> int:
+        """Return the length of the main task queue."""
+        pass
+
+    @abstractmethod
+    def cleanup_dlq(self) -> None:
+        """Clear all items from dead letter queue."""
+        pass

modelq/app/backends/redis/__init__.py

@@ -0,0 +1,3 @@
+from modelq.app.backends.redis.backend import RedisQueueBackend
+
+__all__ = ["RedisQueueBackend"]

modelq/app/backends/redis/backend.py

@@ -0,0 +1,138 @@
+import time
+import json
+import redis
+from typing import Optional
+from modelq.app.backends.base import QueueBackend
+
+
+class RedisQueueBackend(QueueBackend):
+    def __init__(self, redis_client: redis.Redis):
+        self.redis = redis_client
+
+    # ─────────────────────── Task Queue ───────────────────────────────
+
+    def enqueue_task(self, task_data: dict) -> None:
+        task_data["status"] = "queued"
+        self.redis.rpush("ml_tasks", json.dumps(task_data))
+        self.redis.zadd("queued_requests", {task_data["task_id"]: task_data["queued_at"]})
+
+    def dequeue_task(self, timeout: Optional[int] = None) -> Optional[dict]:
+        data = self.redis.blpop("ml_tasks", timeout=timeout or 5)
+        if data:
+            _, task_json = data
+            return json.loads(task_json)
+        return None
+
+    def requeue_task(self, task_data: dict) -> None:
+        self.redis.rpush("ml_tasks", json.dumps(task_data))
+
+    def enqueue_delayed_task(self, task_data: dict, delay_seconds: int) -> None:
+        run_at = time.time() + delay_seconds
+        self.redis.zadd("delayed_tasks", {json.dumps(task_data): run_at})
+
+    def dequeue_ready_delayed_tasks(self) -> list:
+        now = time.time()
+        tasks = self.redis.zrangebyscore("delayed_tasks", 0, now)
+        for task_json in tasks:
+            self.redis.zrem("delayed_tasks", task_json)
+            self.redis.lpush("ml_tasks", task_json)
+        return [json.loads(t) for t in tasks]
+
+    def flush_queue(self) -> None:
+        self.redis.ltrim("ml_tasks", 1, 0)
+
+    # ─────────────────────── Task State ───────────────────────────────
+
+    def save_task_state(self, task_id: str, task_data: dict, result: bool) -> None:
+        task_data["finished_at"] = time.time()
+        self.redis.set(f"task_result:{task_id}", json.dumps(task_data), ex=3600)
+        self.redis.set(f"task:{task_id}", json.dumps(task_data), ex=86400)
+
+    def load_task_state(self, task_id: str) -> Optional[dict]:
+        data = self.redis.get(f"task:{task_id}")
+        return json.loads(data) if data else None
+
+    def remove_task_from_queue(self, task_id: str) -> bool:
+        tasks = self.redis.lrange("ml_tasks", 0, -1)
+        for task_json in tasks:
+            task_dict = json.loads(task_json)
+            if task_dict.get("task_id") == task_id:
+                self.redis.lrem("ml_tasks", 1, task_json)
+                self.redis.zrem("queued_requests", task_id)
+                return True
+        return False
+
+    def mark_processing(self, task_id: str) -> bool:
+        return self.redis.sadd("processing_tasks", task_id) == 1
+
+    def unmark_processing(self, task_id: str) -> None:
+        self.redis.srem("processing_tasks", task_id)
+
+    def get_all_processing_tasks(self) -> list:
+        return [pid.decode() for pid in self.redis.smembers("processing_tasks")]
+
+    def get_all_queued_tasks(self) -> list:
+        raw = self.redis.lrange("ml_tasks", 0, -1)
+        return [json.loads(task) for task in raw if json.loads(task).get("status") == "queued"]
+
+    # ─────────────────────── Server State ───────────────────────────────
+
+    def register_server(self, server_id: str, task_names: list) -> None:
+        self.redis.hset("servers", server_id, json.dumps({
+            "allowed_tasks": task_names,
+            "status": "idle",
+            "last_heartbeat": time.time()
+        }))
+
+    def update_server_status(self, server_id: str, status: str) -> None:
+        raw = self.redis.hget("servers", server_id)
+        if raw:
+            data = json.loads(raw)
+            data["status"] = status
+            data["last_heartbeat"] = time.time()
+            self.redis.hset("servers", server_id, json.dumps(data))
+
+    def get_all_server_ids(self) -> list:
+        return [k.decode("utf-8") for k in self.redis.hkeys("servers")]
+
+    def get_server_data(self, server_id: str) -> Optional[dict]:
+        raw = self.redis.hget("servers", server_id)
+        return json.loads(raw) if raw else None
+
+    def prune_dead_servers(self, timeout: int) -> list:
+        now = time.time()
+        pruned = []
+        for sid, raw in self.redis.hgetall("servers").items():
+            try:
+                sid_str = sid.decode()
+                data = json.loads(raw)
+                if now - data.get("last_heartbeat", 0) > timeout:
+                    self.redis.hdel("servers", sid_str)
+                    pruned.append(sid_str)
+            except:
+                continue
+        return pruned
+
+    # ─────────────────────── Miscellaneous ─────────────────────────────
+
+    def prune_old_results(self, older_than_seconds: int) -> int:
+        now = time.time()
+        deleted = 0
+        for key in self.redis.scan_iter("task_result:*"):
+            raw = self.redis.get(key)
+            if not raw:
+                continue
+            data = json.loads(raw)
+            timestamp = data.get("finished_at") or data.get("started_at")
+            if timestamp and now - timestamp > older_than_seconds:
+                task_id = key.decode().split(":")[-1]
+                self.redis.delete(key)
+                self.redis.delete(f"task:{task_id}")
+                deleted += 1
+        return deleted
+
+    def queue_length(self) -> int:
+        return self.redis.llen("ml_tasks")
+
+    def cleanup_dlq(self) -> None:
+        self.redis.delete("dlq")