Keep public statistics available during source commits

src/fortune_intel/api.py

@@ -2,7 +2,11 @@
 
 from __future__ import annotations
 
+import sqlite3
 from pathlib import Path
+from threading import Lock
+from time import monotonic
+from typing import Any
 from xml.sax.saxutils import escape as xml_escape
 
 from fastapi import FastAPI, HTTPException, Query
@@ -24,6 +28,7 @@
 )
 
 COUNTRY_SCOPE = "United States (50 states and Washington, DC)"
+_STATS_CACHE_SECONDS = 300.0
 
 
 def create_app(
@@ -44,6 +49,9 @@ def create_app(
     )
     app.state.repository = repository
     app.state.settings = settings
+    stats_cache: dict[str, Any] | None = None
+    stats_cache_at = 0.0
+    stats_cache_lock = Lock()
     app.add_middleware(TrustedHostMiddleware, allowed_hosts=list(settings.allowed_hosts))
     if settings.cors_origins:
         app.add_middleware(
@@ -75,9 +83,25 @@ def readiness() -> dict[str, object]:
 
     @app.get("/api/stats")
     def stats() -> dict[str, object]:
-        result = repository.overview(include_synthetic=settings.show_synthetic, us_only=True)
-        result["h1b_employers"] = repository.h1b_overview()["employers"]
-        result["country_scope"] = COUNTRY_SCOPE
+        nonlocal stats_cache, stats_cache_at
+        now = monotonic()
+        with stats_cache_lock:
+            if stats_cache is not None and now - stats_cache_at < _STATS_CACHE_SECONDS:
+                return dict(stats_cache)
+        try:
+            result = repository.overview(include_synthetic=settings.show_synthetic, us_only=True)
+            result["h1b_employers"] = repository.h1b_overview()["employers"]
+            result["country_scope"] = COUNTRY_SCOPE
+        except sqlite3.OperationalError as error:
+            with stats_cache_lock:
+                if stats_cache is not None:
+                    return dict(stats_cache)
+            raise HTTPException(
+                status_code=503, detail="statistics temporarily unavailable"
+            ) from error
+        with stats_cache_lock:
+            stats_cache = dict(result)
+            stats_cache_at = now
         return result
 
     @app.get("/api/coverage")

tests/test_api.py

@@ -1,3 +1,5 @@
+import sqlite3
+
 from fastapi.testclient import TestClient
 
 from fortune_intel.api import create_app
@@ -22,6 +24,23 @@ def test_search_api_and_stats(tmp_path):
     assert stats["jobs_with_evidence"] == 3
 
 
+def test_stats_uses_recent_verified_snapshot_during_a_database_write(tmp_path, monkeypatch):
+    app = create_app(tmp_path / "cached-stats.db")
+    seed_demo(app.state.repository)
+    client = TestClient(app)
+    first = client.get("/api/stats")
+    assert first.status_code == 200
+
+    def locked(**kwargs):
+        raise sqlite3.OperationalError("database is locked")
+
+    monkeypatch.setattr(app.state.repository, "overview", locked)
+    second = client.get("/api/stats")
+
+    assert second.status_code == 200
+    assert second.json() == first.json()
+
+
 def test_invalid_tier_is_rejected(tmp_path):
     client = TestClient(create_app(tmp_path / "api.db"))
     assert client.get("/api/jobs?tier=Z").status_code == 422