Add FastAPI catalog backend (games/snapshots/diff) + tests
Modular-monolith backend over SQLAlchemy (SQLite by default, Postgres-ready
via DATABASE_URL). The full snapshot.json is stored verbatim; diffing reads it
back through the ams.diff engine, so the DB never mirrors the snapshot schema.
- ams.api.db/models/schemas/service : Game 1-N Snapshot, sha256-deduped upsert
- routes: POST/GET /games, POST/GET /snapshots (import, deduped), GET /diff
(?old&new[&owner]) running compute_diff on stored snapshots, /health
- ams.api.importer : bulk CLI loader (python -m ams.api.importer --game ...)
- run: uvicorn ams.api.app:create_app --factory
11 tests pass (6 diff + 5 API via TestClient over the golden pair). Smoke-tested
live on uvicorn: import -> /snapshots -> /diff returns the BlooMoo deltas.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
60
ams/api/service.py
Normal file
60
ams/api/service.py
Normal file
@@ -0,0 +1,60 @@
|
||||
"""Business logic shared by the HTTP routes and the bulk importer."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import models
|
||||
|
||||
|
||||
def _content_sha(data: dict[str, Any]) -> str:
|
||||
return hashlib.sha256(json.dumps(data, sort_keys=True).encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def looks_like_snapshot(data: Any) -> bool:
|
||||
return isinstance(data, dict) and "binary" in data and "types" in data
|
||||
|
||||
|
||||
def _apply_metadata(snap: models.Snapshot, data: dict[str, Any]) -> None:
|
||||
binary = data.get("binary", {})
|
||||
snap.binary_name = binary.get("name", "?")
|
||||
snap.engine = binary.get("engine")
|
||||
snap.compiler = binary.get("compiler")
|
||||
snap.schema_version = data.get("schema_version")
|
||||
snap.n_types = len(data.get("types", []))
|
||||
snap.n_methods = len(data.get("methods", []))
|
||||
snap.n_events = len(data.get("events", []))
|
||||
snap.n_fields = len(data.get("fields", []))
|
||||
snap.data = data
|
||||
|
||||
|
||||
def _get_or_create_game(db: Session, name: str) -> models.Game:
|
||||
game = db.scalar(select(models.Game).where(models.Game.name == name))
|
||||
if game is None:
|
||||
game = models.Game(name=name)
|
||||
db.add(game)
|
||||
db.flush()
|
||||
return game
|
||||
|
||||
|
||||
def import_snapshot(db: Session, data: dict[str, Any], game_name: str | None = None) -> models.Snapshot:
|
||||
"""Upsert a snapshot, deduped by the binary's sha256 (falling back to a content hash)."""
|
||||
sha = data.get("binary", {}).get("sha256") or _content_sha(data)
|
||||
snap = db.scalar(select(models.Snapshot).where(models.Snapshot.sha256 == sha))
|
||||
game = _get_or_create_game(db, game_name) if game_name else None
|
||||
|
||||
if snap is None:
|
||||
snap = models.Snapshot(sha256=sha)
|
||||
db.add(snap)
|
||||
_apply_metadata(snap, data)
|
||||
if game is not None:
|
||||
snap.game_id = game.id
|
||||
|
||||
db.commit()
|
||||
db.refresh(snap)
|
||||
return snap
|
||||
Reference in New Issue
Block a user