"""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