diff --git a/pipeline/dags/school_data_pipeline.py b/pipeline/dags/school_data_pipeline.py index b8f85da..79529ab 100644 --- a/pipeline/dags/school_data_pipeline.py +++ b/pipeline/dags/school_data_pipeline.py @@ -38,6 +38,30 @@ 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}" \\ + -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 +115,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 +150,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 +187,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) ────────────────────────────────────