8386196653
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>
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
"""Bulk-import snapshot.json files straight into the DB (no HTTP server needed).
|
|
|
|
python -m ams.api.importer [--game "Reksio i UFO"] snapshots/*.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
|
|
from .db import get_session, init_db
|
|
from .service import import_snapshot, looks_like_snapshot
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
p = argparse.ArgumentParser(prog="ams-import", description="Import snapshots into the catalog DB.")
|
|
p.add_argument("files", nargs="+", help="snapshot.json files")
|
|
p.add_argument("--game", help="link all imported snapshots to this game (created if missing)")
|
|
args = p.parse_args(argv)
|
|
|
|
init_db()
|
|
db = get_session()
|
|
try:
|
|
for path in args.files:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
if not looks_like_snapshot(data):
|
|
print("[!] skip (not a snapshot): {0}".format(path))
|
|
continue
|
|
snap = import_snapshot(db, data, args.game)
|
|
print("[+] #{0} {1} [{2}/{3}] types={4} methods={5} events={6} fields={7}".format(
|
|
snap.id, snap.binary_name, snap.engine, snap.compiler,
|
|
snap.n_types, snap.n_methods, snap.n_events, snap.n_fields))
|
|
finally:
|
|
db.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|