135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
"""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
|
|
|