Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d677b54533 | ||
|
|
c353e36072 |
@@ -38,6 +38,31 @@ 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}" \\
|
||||||
|
--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) ──────────────────────────────────────
|
# ── 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",
|
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 +151,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 +188,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) ────────────────────────────────────
|
||||||
@@ -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+",
|
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
|
||||||
|
|||||||
Reference in New Issue
Block a user