fix(pipeline): cache invalidation for IDACI DAG too; curl timeouts

Addresses AI-review findings: the annual IDACI DAG also rebuilds a mart
(fact_deprivation) and needs the reload; curl gets connect/max timeouts
so an unreachable backend fails fast instead of hanging the task.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/data_loader.py

@@ -262,53 +262,15 @@ assert "NULL AS has_sixth_form" in str(_MAIN_QUERY_NO_SIXTH_FORM), (
     "expected replacement of 's.has_sixth_form,' to have taken effect"
 )
 
-# Fallback used when marts.dim_school predates the GIAS code-dictionary
-# migration (i.e. the nightly dbt pipeline hasn't rebuilt the mart yet on
-# this DB, so it still has the old name columns instead of *_code columns).
-_MAIN_QUERY_LEGACY_NAMES = str(_MAIN_QUERY)
-_LEGACY_NAME_REPLACEMENTS = [
-    ("s.phase_code,", "s.phase,"),
-    ("s.school_type_code,", "s.school_type,"),
-    (
-        "s.religious_character_code,",
-        "s.religious_character AS religious_denomination,",
-    ),
-    ("s.status_code,", "s.status,"),
-    ("s.admissions_policy_code,", "s.admissions_policy,"),
-]
-for _old, _new in _LEGACY_NAME_REPLACEMENTS:
-    assert _old in _MAIN_QUERY_LEGACY_NAMES, (
-        f"expected {_old!r} to be present in _MAIN_QUERY before replacement"
-    )
-    _MAIN_QUERY_LEGACY_NAMES = _MAIN_QUERY_LEGACY_NAMES.replace(_old, _new)
-_MAIN_QUERY_LEGACY_NAMES = text(_MAIN_QUERY_LEGACY_NAMES)
-
-_GIAS_CODE_COLUMN_NAMES = (
-    "phase_code",
-    "school_type_code",
-    "religious_character_code",
-    "status_code",
-    "admissions_policy_code",
-)
-
 
 def load_school_data_as_dataframe() -> pd.DataFrame:
     """Load all school + KS2 data as a pandas DataFrame."""
     try:
         df = pd.read_sql(_MAIN_QUERY, engine)
     except sqlalchemy.exc.ProgrammingError as exc:
-        if any(col in str(exc) for col in _GIAS_CODE_COLUMN_NAMES):
+        if "has_sixth_form" not in str(exc):
-            logging.getLogger(__name__).warning(
+            print(f"Warning: Could not load school data from marts: {exc}")
-                "marts predate the GIAS code migration — falling back to "
-                "legacy name-column query: %s",
-                exc,
-            )
-            try:
-                df = pd.read_sql(_MAIN_QUERY_LEGACY_NAMES, engine)
-            except Exception as exc2:
-                print(f"Warning: Could not load school data from marts: {exc2}")
             return pd.DataFrame()
-        elif "has_sixth_form" in str(exc):
         logging.getLogger(__name__).warning(
             "marts.dim_school is missing has_sixth_form (pipeline hasn't "
             "rebuilt the mart yet on this DB) — retrying without it: %s",
@@ -319,9 +281,6 @@ def load_school_data_as_dataframe() -> pd.DataFrame:
         except Exception as exc2:
             print(f"Warning: Could not load school data from marts: {exc2}")
             return pd.DataFrame()
-        else:
-            print(f"Warning: Could not load school data from marts: {exc}")
-            return pd.DataFrame()
     except Exception as exc:
         print(f"Warning: Could not load school data from marts: {exc}")
         return pd.DataFrame()

backend/tests/test_gias_translation.py

@@ -42,55 +42,3 @@ def test_missing_code_columns_are_a_noop():
     out = translate_gias_code_columns(df)
     assert out.iloc[0]["phase"] == "Primary"
     assert out.iloc[0]["status"] == "Open"
-
-
-def test_load_school_data_survives_premigration_marts(monkeypatch):
-    """Real prod state until the nightly pipeline first rebuilds the mart with
-    the GIAS code columns: marts.dim_school still has the old name columns
-    (phase, school_type, religious_character, status, admissions_policy)
-    instead of the new *_code columns. The first query raises UndefinedColumn
-    on s.phase_code; load_school_data_as_dataframe must retry with the
-    legacy name-column query rather than swallow the error and return (and
-    then have load_school_data cache) an empty DataFrame."""
-    import sqlalchemy.exc
-    from backend import data_loader
-
-    data_loader._df_cache = None
-    data_loader._df_latest_cache = None
-
-    good_df = pd.DataFrame(
-        [
-            {
-                "urn": 1,
-                "school_name": "Legacy School",
-                "phase": "Primary",
-                "school_type": "Academy",
-                "status": "Open",
-            }
-        ]
-    )
-    calls = []
-
-    def fake_read_sql(query, con):
-        calls.append(query)
-        if len(calls) == 1:
-            raise sqlalchemy.exc.ProgrammingError(
-                "(psycopg2.errors.UndefinedColumn) column s.phase_code does not exist",
-                None,
-                None,
-            )
-        return good_df.copy()
-
-    monkeypatch.setattr(data_loader.pd, "read_sql", fake_read_sql)
-
-    try:
-        df = data_loader.load_school_data_as_dataframe()
-    finally:
-        data_loader._df_cache = None
-        data_loader._df_latest_cache = None
-
-    assert len(calls) == 2, "must retry with the legacy name-column query variant"
-    assert calls[1] is data_loader._MAIN_QUERY_LEGACY_NAMES
-    assert not df.empty
-    assert df["phase"].iloc[0] == "Primary"
-    assert df["status"].iloc[0] == "Open"

pipeline/dags/school_data_pipeline.py

@@ -38,6 +38,31 @@ default_args = {
     "retry_delay": timedelta(minutes=5),
 }
 
+# The backend caches the marts DataFrame at startup; after any rebuild the
+# cache must be invalidated or the API serves stale (or empty) data until the
+# container restarts.
+INVALIDATE_CACHE_CMD = """
+set -e
+BACKEND_URL="${BACKEND_URL:-http://backend:80}"
+ADMIN_KEY="${ADMIN_API_KEY:-changeme}"
+
+echo "Calling $BACKEND_URL/api/admin/reload ..."
+
+response=$(curl -s -o /tmp/reload_response.json -w "%{http_code}" \\
+    --connect-timeout 10 --max-time 120 \\
+    -X POST "$BACKEND_URL/api/admin/reload" \\
+    -H "X-API-Key: $ADMIN_KEY" \\
+    -H "Content-Type: application/json")
+
+echo "HTTP status: $response"
+cat /tmp/reload_response.json
+
+if [ "$response" != "200" ]; then
+    echo "ERROR: backend cache reload failed (HTTP $response)"
+    exit 1
+fi
+"""
+
 
 # ── Daily DAG (GIAS + downstream) ──────────────────────────────────────
 
@@ -91,7 +116,12 @@ print(f'Validation passed: {{count}} GIAS rows')
         bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
     )
 
-    extract_group >> validate_raw >> dbt_build >> sync_typesense
+    invalidate_cache = BashOperator(
+        task_id="invalidate_cache",
+        bash_command=INVALIDATE_CACHE_CMD,
+    )
+
+    extract_group >> validate_raw >> dbt_build >> sync_typesense >> invalidate_cache
 
 
 # ── Monthly DAG (Ofsted) ───────────────────────────────────────────────
@@ -121,7 +151,12 @@ with DAG(
         bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
     )
 
-    extract_ofsted >> dbt_build_ofsted >> sync_typesense_ofsted
+    invalidate_cache_ofsted = BashOperator(
+        task_id="invalidate_cache",
+        bash_command=INVALIDATE_CACHE_CMD,
+    )
+
+    extract_ofsted >> dbt_build_ofsted >> sync_typesense_ofsted >> invalidate_cache_ofsted
 
 
 # ── Annual DAG (EES: KS2, KS4, Census, Admissions) ───────────────────
@@ -153,7 +188,12 @@ with DAG(
         bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
     )
 
-    extract_ees_group >> dbt_build_ees >> sync_typesense_ees
+    invalidate_cache_ees = BashOperator(
+        task_id="invalidate_cache",
+        bash_command=INVALIDATE_CACHE_CMD,
+    )
+
+    extract_ees_group >> dbt_build_ees >> sync_typesense_ees >> invalidate_cache_ees
 
 
 # ── Annual DAG (IDACI Deprivation) ────────────────────────────────────
@@ -178,4 +218,9 @@ with DAG(
         bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_idaci+ fact_deprivation+",
     )
 
-    extract_idaci >> dbt_build_idaci
+    invalidate_cache_idaci = BashOperator(
+        task_id="invalidate_cache",
+        bash_command=INVALIDATE_CACHE_CMD,
+    )
+
+    extract_idaci >> dbt_build_idaci >> invalidate_cache_idaci