fix(pipeline): actually invalidate the backend cache after data rebuilds
PR Checks / Frontend Typecheck + Tests (pull_request) Successful in 9m40s
PR Checks / Backend Smoke (pull_request) Successful in 6s
PR Checks / Build Backend (no push) (pull_request) Successful in 16s
PR Checks / Build Frontend (no push) (pull_request) Successful in 51s
PR Checks / Build Pipeline (no push) (pull_request) Successful in 36s
PR Checks / AI Code Review (Claude) (pull_request) Failing after 1m20s

The daily/monthly/annual DAG docstring promised an Invalidate Cache step
that never existed — after a marts rebuild the backend kept serving its
startup-cached (possibly empty) DataFrame until a container restart.
Add a POST /api/admin/reload task at the end of each pipeline DAG,
mirroring the sitemap DAG's admin-call pattern.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Tudor
2026-07-09 21:12:01 +01:00
co-authored by Claude Fable 5
parent 84dfc6c1bb
commit c353e36072
+42 -3
View File
@@ -38,6 +38,30 @@ default_args = {
"retry_delay": timedelta(minutes=5), "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}" \\
-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) ────────────────────────────────────── # ── Daily DAG (GIAS + downstream) ──────────────────────────────────────
@@ -91,7 +115,12 @@ print(f'Validation passed: {{count}} GIAS rows')
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py", 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) ─────────────────────────────────────────────── # ── Monthly DAG (Ofsted) ───────────────────────────────────────────────
@@ -121,7 +150,12 @@ with DAG(
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py", 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) ─────────────────── # ── Annual DAG (EES: KS2, KS4, Census, Admissions) ───────────────────
@@ -153,7 +187,12 @@ with DAG(
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py", 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) ──────────────────────────────────── # ── Annual DAG (IDACI Deprivation) ────────────────────────────────────