perf: cache media manifest revisions

apps/api/app/media_catalog.py

@@ -11,6 +11,22 @@ WATERBODY_MEDIA_ROLES = {"waterbody_cover", "waterbody_map", "waterbody_depth_ma
 TACKLE_MEDIA_ROLES = {"tackle_card", "tackle_detail", "rig_diagram", "tackle_screenshot"}
 KNOWN_MEDIA_ROLES = WATERBODY_MEDIA_ROLES | TACKLE_MEDIA_ROLES
 MEDIA_ROLES_BY_ENTITY = {"waterbody": WATERBODY_MEDIA_ROLES, "tackle": TACKLE_MEDIA_ROLES}
+_manifest_cache: tuple[Path, int, int, dict] | None = None
+
+
+def _read_manifest() -> dict:
+    """Read the manifest once per file revision in this API process."""
+    global _manifest_cache
+    path = (MEDIA_ROOT / "manifest.json").resolve()
+    stat = path.stat()
+    marker = (path, stat.st_mtime_ns, stat.st_size)
+    if _manifest_cache and _manifest_cache[:3] == marker:
+        return _manifest_cache[3]
+    manifest = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(manifest, dict):
+        raise ValueError("media manifest must be an object")
+    _manifest_cache = (*marker, manifest)
+    return manifest
 
 
 def _source_system(source_page: object) -> str:
@@ -30,7 +46,7 @@ def _source_system(source_page: object) -> str:
 
 
 def media_manifest_version() -> int:
-    manifest = json.loads((MEDIA_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest = _read_manifest()
     return max(1, int(manifest.get("version", 1)))
 
 
@@ -47,7 +63,7 @@ def _public_role_allowed(entity_type: str | None, role: object) -> bool:
 
 
 def published_assets(entity_type: str | None = None, media_role: str | None = None) -> list[dict]:
-    manifest = json.loads((MEDIA_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest = _read_manifest()
     result = []
     for item in manifest.get("assets", []):
         if item.get("status") != "approved" or not item.get("sha256") or not item.get("local_path"):
@@ -90,7 +106,7 @@ def published_assets(entity_type: str | None = None, media_role: str | None = No
 def published_file(digest: str) -> tuple[Path, str] | None:
     if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
         return None
-    manifest = json.loads((MEDIA_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest = _read_manifest()
     item = next((row for row in manifest.get("assets", []) if row.get("status") == "approved" and row.get("sha256") == digest), None)
     media_type = None
     local_path = None
@@ -118,7 +134,7 @@ def review_assets(
     entity_type: str | None = None, status: str | None = None, media_role: str | None = None,
     query: str | None = None,
 ) -> list[dict]:
-    manifest = json.loads((MEDIA_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest = _read_manifest()
     result = []
     for item in manifest.get("assets", []):
         item_status = str(item.get("status") or "")
@@ -161,7 +177,7 @@ def review_assets(
 def review_file(digest: str) -> tuple[Path, str] | None:
     if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
         return None
-    manifest = json.loads((MEDIA_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest = _read_manifest()
     item = next((row for row in manifest.get("assets", []) if row.get("sha256") == digest and row.get("status") in {"approved", "upgrade_queued", "upgrade_stored"}), None)
     if not item or not item.get("local_path"):
         return None