feat(03-01): SwipeService CRUD + Pydantic schemas + FastAPI router

- backend/schemas/swipe.py: SwipeItem, SwipeItemCreate, SwipeItemUpdate, SwipeListResponse
- backend/services/swipe_service.py: SwipeService con load/save/add/update/delete/mark_used su swipe_file.json
- backend/routers/swipe.py: 5 endpoint REST (GET/POST/PUT/DELETE/mark-used), lazy init pattern
- backend/main.py: registra swipe.router prima del SPA catch-all mount
This commit is contained in:
Michele
2026-03-09 00:22:12 +01:00
parent 05988f4be5
commit d64c7f4524
4 changed files with 376 additions and 1 deletions

View File

@@ -0,0 +1,181 @@
"""SwipeService — CRUD per il file data/swipe_file.json.
Responsabilita':
- Carica e salva le voci dello Swipe File in un file JSON su disco.
- Ordina le voci per data creazione (piu' recenti prima).
- Genera ID brevi (uuid4.hex[:12]) e timestamp ISO UTC.
- Espone add/update/delete/mark_used come operazioni atomiche.
Nota: le operazioni di I/O sono sincrone (json.load / json.dump).
Il file e' semplice JSON (lista di dict) e non richiede lock
perche' FastAPI e' single-process per questo caso d'uso.
"""
from __future__ import annotations
import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from backend.schemas.swipe import SwipeItem, SwipeItemCreate, SwipeItemUpdate
class SwipeService:
"""Gestisce la persistenza del file swipe_file.json."""
def __init__(self, data_path: Path) -> None:
"""
Args:
data_path: Directory DATA_PATH. Il file sara' creato come
data_path / "swipe_file.json".
"""
self._file = data_path / "swipe_file.json"
# ------------------------------------------------------------------
# Private helpers
# ------------------------------------------------------------------
def _load(self) -> list[dict]:
"""Carica il file JSON dal disco.
Returns:
Lista di dict. Ritorna lista vuota se il file non esiste.
"""
if not self._file.exists():
return []
try:
return json.loads(self._file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return []
def _save(self, items: list[dict]) -> None:
"""Salva la lista di dict nel file JSON.
Args:
items: Lista di dict da persistere.
"""
self._file.write_text(
json.dumps(items, ensure_ascii=False, indent=2),
encoding="utf-8",
)
@staticmethod
def _now_iso() -> str:
"""Ritorna timestamp corrente in formato ISO 8601 UTC."""
return datetime.now(timezone.utc).isoformat()
@staticmethod
def _new_id() -> str:
"""Genera un ID breve (12 caratteri hex)."""
return uuid.uuid4().hex[:12]
# ------------------------------------------------------------------
# Public interface
# ------------------------------------------------------------------
def list_items(self) -> list[SwipeItem]:
"""Ritorna tutte le voci ordinate per data creazione (piu' recenti prima).
Returns:
Lista di SwipeItem ordinata cronologicamente inversa.
"""
items = self._load()
# Ordina per created_at descending (ISO string sort works correctly)
items.sort(key=lambda x: x.get("created_at", ""), reverse=True)
return [SwipeItem(**item) for item in items]
def add_item(self, data: SwipeItemCreate) -> SwipeItem:
"""Aggiunge una nuova voce allo Swipe File.
Args:
data: Dati validati per la nuova voce.
Returns:
SwipeItem creato con id e timestamp generati.
"""
now = self._now_iso()
item = {
"id": self._new_id(),
"topic": data.topic,
"nicchia": data.nicchia,
"note": data.note,
"created_at": now,
"updated_at": now,
"used": False,
}
items = self._load()
items.append(item)
self._save(items)
return SwipeItem(**item)
def update_item(self, item_id: str, data: SwipeItemUpdate) -> SwipeItem:
"""Aggiorna una voce esistente.
Aggiorna solo i campi non-None presenti in data.
Aggiorna sempre updated_at.
Args:
item_id: ID della voce da aggiornare.
data: Campi da aggiornare.
Returns:
SwipeItem aggiornato.
Raises:
ValueError: Se l'ID non esiste.
"""
items = self._load()
for item in items:
if item["id"] == item_id:
if data.topic is not None:
item["topic"] = data.topic
if data.nicchia is not None:
item["nicchia"] = data.nicchia
if data.note is not None:
item["note"] = data.note
item["updated_at"] = self._now_iso()
self._save(items)
return SwipeItem(**item)
raise ValueError(f"SwipeItem con id='{item_id}' non trovato")
def delete_item(self, item_id: str) -> bool:
"""Elimina una voce dallo Swipe File.
Args:
item_id: ID della voce da eliminare.
Returns:
True se eliminata con successo.
Raises:
ValueError: Se l'ID non esiste.
"""
items = self._load()
original_len = len(items)
items = [item for item in items if item["id"] != item_id]
if len(items) == original_len:
raise ValueError(f"SwipeItem con id='{item_id}' non trovato")
self._save(items)
return True
def mark_used(self, item_id: str) -> SwipeItem:
"""Segna una voce come utilizzata in un calendario.
Args:
item_id: ID della voce da segnare come usata.
Returns:
SwipeItem aggiornato con used=True.
Raises:
ValueError: Se l'ID non esiste.
"""
items = self._load()
for item in items:
if item["id"] == item_id:
item["used"] = True
item["updated_at"] = self._now_iso()
self._save(items)
return SwipeItem(**item)
raise ValueError(f"SwipeItem con id='{item_id}' non trovato")