tiphys.scheduler.store

  1"""
  2Task store for scheduled tasks.
  3
  4Persists tasks to disk as JSON files with file locking.
  5"""
  6
  7import contextlib
  8import json
  9from datetime import datetime
 10from pathlib import Path
 11
 12import structlog
 13from filelock import FileLock
 14
 15from tiphys.scheduler.models import ScheduledTask
 16
 17logger = structlog.get_logger(__name__)
 18
 19
 20class TaskStore:
 21    """
 22    Persistent store for scheduled tasks.
 23
 24    Uses file locking for cross-process safety.
 25    """
 26
 27    def __init__(self, tasks_dir: Path | None = None):
 28        """
 29        Initialize the task store.
 30
 31        Args:
 32            tasks_dir: Directory for task files.
 33                      Defaults to ~/.tiphys/tasks/
 34        """
 35        if tasks_dir is None:
 36            tasks_dir = Path.home() / ".tiphys" / "tasks"
 37        self.tasks_dir = tasks_dir
 38        self.tasks_dir.mkdir(parents=True, exist_ok=True)
 39
 40    def _task_path(self, task_id: str) -> Path:
 41        """Get path to task file."""
 42        return self.tasks_dir / f"{task_id}.json"
 43
 44    def _lock_path(self, task_id: str) -> Path:
 45        """Get path to lock file."""
 46        return self._task_path(task_id).with_suffix(".lock")
 47
 48    def save(self, task: ScheduledTask) -> None:
 49        """Save a task to disk."""
 50        task.updated_at = datetime.utcnow()
 51        path = self._task_path(task.id)
 52
 53        lock = FileLock(self._lock_path(task.id))
 54        with lock:
 55            temp_path = path.with_suffix(".tmp")
 56            temp_path.write_text(task.model_dump_json(indent=2))
 57            temp_path.rename(path)
 58
 59    def load(self, task_id: str) -> ScheduledTask | None:
 60        """Load a task from disk."""
 61        path = self._task_path(task_id)
 62        if not path.exists():
 63            return None
 64
 65        lock = FileLock(self._lock_path(task_id))
 66        with lock:
 67            try:
 68                data = json.loads(path.read_text())
 69                return ScheduledTask.model_validate(data)
 70            except (json.JSONDecodeError, FileNotFoundError):
 71                return None
 72
 73    def delete(self, task_id: str) -> bool:
 74        """Delete a task from disk."""
 75        path = self._task_path(task_id)
 76        lock_path = self._lock_path(task_id)
 77
 78        if not path.exists():
 79            return False
 80
 81        lock = FileLock(lock_path)
 82        with lock:
 83            try:
 84                path.unlink()
 85                with contextlib.suppress(FileNotFoundError):
 86                    lock_path.unlink()
 87                return True
 88            except FileNotFoundError:
 89                return False
 90
 91    def list_all(self) -> list[ScheduledTask]:
 92        """List all tasks in the store."""
 93        tasks = []
 94        for path in self.tasks_dir.glob("*.json"):
 95            try:
 96                data = json.loads(path.read_text())
 97                tasks.append(ScheduledTask.model_validate(data))
 98            except Exception:
 99                continue
100        return tasks
101
102    def list_by_group(self, group: str) -> list[ScheduledTask]:
103        """List tasks belonging to a specific group."""
104        return [t for t in self.list_all() if t.group == group]
105
106    def get_due_tasks(self, now: datetime | None = None) -> list[ScheduledTask]:
107        """Get all active tasks that are due to run."""
108        if now is None:
109            now = datetime.utcnow()
110
111        return [
112            t for t in self.list_all() if t.status == "active" and t.next_run and t.next_run <= now
113        ]