Spaces:
Running
Running
File size: 2,359 Bytes
2e175db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """
Scan-record persistence.
Important
---------
This module persists ONLY metadata (scan_id, verdict, model_version, latency,
timestamp). The image itself is NEVER stored here. Image storage is a separate,
opt-in path handled by `storage.blob`.
Stage 1 ships an in-memory fallback so the service runs out of the box with
no database. Set DATABASE_URL to switch to Postgres (Neon, Supabase, etc.).
"""
from __future__ import annotations
import logging
import time
from dataclasses import dataclass, field
from ..config import settings
log = logging.getLogger(__name__)
@dataclass
class ScanRecord:
scan_id: str
verdict: str
confidence: float
model_version: str
latency_ms: float
c2pa_present: bool
timestamp_unix: float = field(default_factory=time.time)
# In-memory store — bounded to last N records to avoid leaking memory.
_MEMORY_CAP = 1000
_memory_store: list[ScanRecord] = []
def record_scan(record: ScanRecord) -> None:
"""Persist a scan record. No-op-safe; never raises into the request path."""
try:
if settings.database_url:
_record_postgres(record)
else:
_record_memory(record)
except Exception as exc:
# Persistence failures must NEVER break the user-facing scan response.
log.warning("Failed to persist scan record: %s", exc)
def _record_memory(record: ScanRecord) -> None:
_memory_store.append(record)
if len(_memory_store) > _MEMORY_CAP:
del _memory_store[: len(_memory_store) - _MEMORY_CAP]
def _record_postgres(record: ScanRecord) -> None: # pragma: no cover - infra
"""Lazy import — SQLAlchemy is heavy and not needed in dev/tests."""
# Implementation intentionally deferred until Stage 2 — schema below.
#
# CREATE TABLE scans (
# scan_id TEXT PRIMARY KEY,
# verdict TEXT NOT NULL,
# confidence REAL NOT NULL,
# model_version TEXT NOT NULL,
# latency_ms REAL NOT NULL,
# c2pa_present BOOLEAN NOT NULL,
# created_at TIMESTAMPTZ NOT NULL DEFAULT now()
# );
raise NotImplementedError("Postgres persistence wired up in Stage 2.")
def memory_store_snapshot() -> list[ScanRecord]:
"""Test/debug helper — returns a copy of the in-memory store."""
return list(_memory_store)
|