initial work of the arcline tickets

store.py

@@ -0,0 +1,134 @@
+"""JSON-based data store for ticket persistence."""
+
+import json
+import os
+import threading
+from pathlib import Path
+from typing import Optional
+
+from models import Ticket
+
+DATA_DIR = Path(__file__).parent / "data"
+DATA_FILE = DATA_DIR / "tickets.json"
+
+
+class TicketStore:
+    """Thread-safe JSON-backed ticket store."""
+
+    def __init__(self, filepath: Path = DATA_FILE) -> None:
+        self._filepath = filepath
+        self._lock = threading.Lock()
+        self._next_id = 1
+        self._tickets: dict[int, Ticket] = {}
+        self._load()
+
+    # ── persistence ──────────────────────────────────────────────
+
+    def _load(self) -> None:
+        if not self._filepath.exists():
+            self._save()
+            return
+        with open(self._filepath) as fh:
+            raw: list[dict] = json.load(fh)
+
+        self._tickets.clear()
+        max_id = 0
+        for item in raw:
+            ticket = Ticket.from_dict(item)
+            self._tickets[ticket.id] = ticket
+            if ticket.id and ticket.id > max_id:
+                max_id = ticket.id
+        self._next_id = max_id + 1
+
+    def _save(self) -> None:
+        self._filepath.parent.mkdir(parents=True, exist_ok=True)
+        data = [t.to_dict() for t in sorted(self._tickets.values(), key=lambda t: t.id or 0)]
+        with open(self._filepath, "w") as fh:
+            json.dump(data, fh, indent=2)
+
+    # ── CRUD ─────────────────────────────────────────────────────
+
+    def list_tickets(
+        self,
+        status: Optional[str] = None,
+        priority: Optional[str] = None,
+        search: Optional[str] = None,
+        project: Optional[str] = None,
+    ) -> list[Ticket]:
+        tickets = list(self._tickets.values())
+
+        if status:
+            tickets = [t for t in tickets if t.status.value == status]
+        if priority:
+            tickets = [t for t in tickets if t.priority.value == priority]
+        if project:
+            tickets = [t for t in tickets if t.project.lower() == project.lower()]
+        if search:
+            q = search.lower()
+            tickets = [
+                t
+                for t in tickets
+                if q in t.title.lower()
+                or q in t.description.lower()
+                or q in t.assignee.lower()
+                or q in t.project.lower()
+            ]
+
+        return sorted(tickets, key=lambda t: t.id or 0, reverse=True)
+
+    def get_ticket(self, ticket_id: int) -> Optional[Ticket]:
+        return self._tickets.get(ticket_id)
+
+    def create_ticket(self, ticket: Ticket) -> Ticket:
+        with self._lock:
+            ticket.id = self._next_id
+            self._next_id += 1
+            self._tickets[ticket.id] = ticket
+            self._save()
+        return ticket
+
+    def update_ticket(self, ticket_id: int, updates: dict) -> Optional[Ticket]:
+        ticket = self._tickets.get(ticket_id)
+        if ticket is None:
+            return None
+
+        with self._lock:
+            for key, value in updates.items():
+                if hasattr(ticket, key):
+                    setattr(ticket, key, value)
+            from datetime import datetime, timezone
+
+            ticket.updated_at = datetime.now(timezone.utc).isoformat()
+            self._save()
+        return ticket
+
+    def delete_ticket(self, ticket_id: int) -> bool:
+        with self._lock:
+            if ticket_id in self._tickets:
+                del self._tickets[ticket_id]
+                self._save()
+                return True
+        return False
+
+    def count_by_status(self) -> dict[str, int]:
+        counts: dict[str, int] = {}
+        for t in self._tickets.values():
+            counts[t.status.value] = counts.get(t.status.value, 0) + 1
+        return counts
+
+    def get_projects(self) -> list[str]:
+        """Return a sorted list of unique, non-empty project names."""
+        projects = {t.project.strip() for t in self._tickets.values() if t.project.strip()}
+        return sorted(projects)
+
+
+# Global singleton store
+_store: Optional[TicketStore] = None
+
+
+def get_store() -> TicketStore:
+    global _store
+    if _store is None:
+        _store = TicketStore()
+    return _store
+