Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b44fca902f |
+2
-18
@@ -4,7 +4,6 @@ Provides efficient queries with caching.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import re
|
|
||||||
|
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
import numpy as np
|
import numpy as np
|
||||||
@@ -292,28 +291,13 @@ _GIAS_CODE_COLUMN_NAMES = (
|
|||||||
"admissions_policy_code",
|
"admissions_policy_code",
|
||||||
)
|
)
|
||||||
|
|
||||||
_MISSING_COLUMN_RE = re.compile(r'column "?(?:s\.)?(\w+)"? does not exist')
|
|
||||||
|
|
||||||
|
|
||||||
def _missing_column_name(exc: Exception) -> Optional[str]:
|
|
||||||
"""Name of the missing column from a psycopg2 UndefinedColumn error.
|
|
||||||
|
|
||||||
Inspects exc.orig (the DBAPI error), whose message names only the
|
|
||||||
offending column — str(exc) also embeds the full SQL statement, which
|
|
||||||
contains every column name and therefore must not be matched against.
|
|
||||||
"""
|
|
||||||
orig = getattr(exc, "orig", None)
|
|
||||||
match = _MISSING_COLUMN_RE.search(str(orig) if orig is not None else str(exc))
|
|
||||||
return match.group(1) if match else None
|
|
||||||
|
|
||||||
|
|
||||||
def load_school_data_as_dataframe() -> pd.DataFrame:
|
def load_school_data_as_dataframe() -> pd.DataFrame:
|
||||||
"""Load all school + KS2 data as a pandas DataFrame."""
|
"""Load all school + KS2 data as a pandas DataFrame."""
|
||||||
try:
|
try:
|
||||||
df = pd.read_sql(_MAIN_QUERY, engine)
|
df = pd.read_sql(_MAIN_QUERY, engine)
|
||||||
except sqlalchemy.exc.ProgrammingError as exc:
|
except sqlalchemy.exc.ProgrammingError as exc:
|
||||||
missing = _missing_column_name(exc)
|
if any(col in str(exc) for col in _GIAS_CODE_COLUMN_NAMES):
|
||||||
if missing in _GIAS_CODE_COLUMN_NAMES:
|
|
||||||
logging.getLogger(__name__).warning(
|
logging.getLogger(__name__).warning(
|
||||||
"marts predate the GIAS code migration — falling back to "
|
"marts predate the GIAS code migration — falling back to "
|
||||||
"legacy name-column query: %s",
|
"legacy name-column query: %s",
|
||||||
@@ -324,7 +308,7 @@ def load_school_data_as_dataframe() -> pd.DataFrame:
|
|||||||
except Exception as exc2:
|
except Exception as exc2:
|
||||||
print(f"Warning: Could not load school data from marts: {exc2}")
|
print(f"Warning: Could not load school data from marts: {exc2}")
|
||||||
return pd.DataFrame()
|
return pd.DataFrame()
|
||||||
elif missing == "has_sixth_form":
|
elif "has_sixth_form" in str(exc):
|
||||||
logging.getLogger(__name__).warning(
|
logging.getLogger(__name__).warning(
|
||||||
"marts.dim_school is missing has_sixth_form (pipeline hasn't "
|
"marts.dim_school is missing has_sixth_form (pipeline hasn't "
|
||||||
"rebuilt the mart yet on this DB) — retrying without it: %s",
|
"rebuilt the mart yet on this DB) — retrying without it: %s",
|
||||||
|
|||||||
@@ -78,7 +78,6 @@ OFFICIAL_SIXTH_FORM: dict[int, str] = {
|
|||||||
0: "Not applicable",
|
0: "Not applicable",
|
||||||
1: "Has a sixth form",
|
1: "Has a sixth form",
|
||||||
2: "Does not have a sixth form",
|
2: "Does not have a sixth form",
|
||||||
9: "",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
RELIGIOUS_CHARACTER: dict[int, str] = {
|
RELIGIOUS_CHARACTER: dict[int, str] = {
|
||||||
@@ -129,14 +128,12 @@ RELIGIOUS_CHARACTER: dict[int, str] = {
|
|||||||
47: "Reformed Baptist",
|
47: "Reformed Baptist",
|
||||||
48: "Roman Catholic/Anglican",
|
48: "Roman Catholic/Anglican",
|
||||||
49: "Sunni Deobandi",
|
49: "Sunni Deobandi",
|
||||||
99: "",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
ADMISSIONS_POLICY: dict[int, str] = {
|
ADMISSIONS_POLICY: dict[int, str] = {
|
||||||
0: "Not applicable",
|
0: "Not applicable",
|
||||||
2: "Selective",
|
2: "Selective",
|
||||||
4: "Non-selective",
|
4: "Non-selective",
|
||||||
9: "",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -81,15 +81,3 @@ def test_seed_matches_dictionaries():
|
|||||||
for row in csv.DictReader(fh):
|
for row in csv.DictReader(fh):
|
||||||
seed[row["field"]][int(row["code"])] = row["name"]
|
seed[row["field"]][int(row["code"])] = row["name"]
|
||||||
assert seed == fields
|
assert seed == fields
|
||||||
|
|
||||||
|
|
||||||
def test_blank_name_sentinel_codes_map_to_empty_string():
|
|
||||||
"""GIAS carries codes whose (name) column is blank — e.g. ReligiousCharacter
|
|
||||||
99 (~4k schools) and AdmissionsPolicy 9 (~5.6k schools). The old name
|
|
||||||
pipeline served these as empty strings; the dictionaries must reproduce
|
|
||||||
that ("" is falsy, so UI tag heuristics stay silent) rather than letting
|
|
||||||
them hit the "Unknown (<code>)" path meant for genuinely new codes."""
|
|
||||||
assert RELIGIOUS_CHARACTER[99] == ""
|
|
||||||
assert ADMISSIONS_POLICY[9] == ""
|
|
||||||
assert translate(99, RELIGIOUS_CHARACTER) == ""
|
|
||||||
assert translate(9, ADMISSIONS_POLICY) == ""
|
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ rest of the backend sees must carry today's name strings."""
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
|
|
||||||
from backend.data_loader import _missing_column_name, translate_gias_code_columns
|
from backend.data_loader import translate_gias_code_columns
|
||||||
from backend.gias_codes import ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION
|
from backend.gias_codes import ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION
|
||||||
|
|
||||||
|
|
||||||
@@ -44,39 +44,6 @@ def test_missing_code_columns_are_a_noop():
|
|||||||
assert out.iloc[0]["status"] == "Open"
|
assert out.iloc[0]["status"] == "Open"
|
||||||
|
|
||||||
|
|
||||||
def _fake_exc(orig_message):
|
|
||||||
"""A stand-in for sqlalchemy.exc.ProgrammingError: str(exc) embeds the
|
|
||||||
full SQL statement (deliberately containing every column name below, to
|
|
||||||
prove the matcher doesn't fall back to it), while .orig carries the real
|
|
||||||
DBAPI error message naming only the offending column."""
|
|
||||||
exc = Exception(
|
|
||||||
"SELECT s.phase_code, s.school_type_code, s.religious_character_code, "
|
|
||||||
"s.status_code, s.admissions_policy_code, s.has_sixth_form FROM ... "
|
|
||||||
f"[SQL: ...] (Background on this error at: https://...)"
|
|
||||||
)
|
|
||||||
exc.orig = Exception(orig_message) if orig_message is not None else None
|
|
||||||
return exc
|
|
||||||
|
|
||||||
|
|
||||||
def test_missing_column_name_quoted():
|
|
||||||
assert _missing_column_name(_fake_exc('column "phase_code" does not exist')) == "phase_code"
|
|
||||||
|
|
||||||
|
|
||||||
def test_missing_column_name_unquoted():
|
|
||||||
assert _missing_column_name(_fake_exc("column phase_code does not exist")) == "phase_code"
|
|
||||||
|
|
||||||
|
|
||||||
def test_missing_column_name_table_prefixed():
|
|
||||||
assert (
|
|
||||||
_missing_column_name(_fake_exc("column s.has_sixth_form does not exist"))
|
|
||||||
== "has_sixth_form"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_missing_column_name_no_match_returns_none():
|
|
||||||
assert _missing_column_name(_fake_exc("relation \"marts.dim_school\" does not exist")) is None
|
|
||||||
|
|
||||||
|
|
||||||
def test_load_school_data_survives_premigration_marts(monkeypatch):
|
def test_load_school_data_survives_premigration_marts(monkeypatch):
|
||||||
"""Real prod state until the nightly pipeline first rebuilds the mart with
|
"""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
|
the GIAS code columns: marts.dim_school still has the old name columns
|
||||||
@@ -108,12 +75,9 @@ def test_load_school_data_survives_premigration_marts(monkeypatch):
|
|||||||
calls.append(query)
|
calls.append(query)
|
||||||
if len(calls) == 1:
|
if len(calls) == 1:
|
||||||
raise sqlalchemy.exc.ProgrammingError(
|
raise sqlalchemy.exc.ProgrammingError(
|
||||||
statement=str(data_loader._MAIN_QUERY),
|
"(psycopg2.errors.UndefinedColumn) column s.phase_code does not exist",
|
||||||
params=None,
|
None,
|
||||||
orig=Exception(
|
None,
|
||||||
"(psycopg2.errors.UndefinedColumn) column s.phase_code "
|
|
||||||
"does not exist\nLINE 5: s.phase_code,"
|
|
||||||
),
|
|
||||||
)
|
)
|
||||||
return good_df.copy()
|
return good_df.copy()
|
||||||
|
|
||||||
|
|||||||
@@ -148,14 +148,10 @@ def test_load_school_data_survives_missing_has_sixth_form_column(monkeypatch):
|
|||||||
def fake_read_sql(query, con):
|
def fake_read_sql(query, con):
|
||||||
calls.append(query)
|
calls.append(query)
|
||||||
if len(calls) == 1:
|
if len(calls) == 1:
|
||||||
# The statement text still contains phase_code, school_type_code,
|
|
||||||
# etc. (it's the full _MAIN_QUERY SELECT list) — that's exactly
|
|
||||||
# the collision this test guards against: matching must be done
|
|
||||||
# against exc.orig (the DBAPI error), not str(exc)/the statement.
|
|
||||||
raise sqlalchemy.exc.ProgrammingError(
|
raise sqlalchemy.exc.ProgrammingError(
|
||||||
statement=str(data_loader._MAIN_QUERY),
|
"SELECT ...",
|
||||||
params=None,
|
None,
|
||||||
orig=Exception(
|
Exception(
|
||||||
"(psycopg2.errors.UndefinedColumn) column s.has_sixth_form "
|
"(psycopg2.errors.UndefinedColumn) column s.has_sixth_form "
|
||||||
"does not exist"
|
"does not exist"
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -38,31 +38,6 @@ 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) ──────────────────────────────────────
|
||||||
|
|
||||||
@@ -116,12 +91,7 @@ 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",
|
||||||
)
|
)
|
||||||
|
|
||||||
invalidate_cache = BashOperator(
|
extract_group >> validate_raw >> dbt_build >> sync_typesense
|
||||||
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) ───────────────────────────────────────────────
|
||||||
@@ -151,12 +121,7 @@ with DAG(
|
|||||||
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
|
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
|
||||||
)
|
)
|
||||||
|
|
||||||
invalidate_cache_ofsted = BashOperator(
|
extract_ofsted >> dbt_build_ofsted >> sync_typesense_ofsted
|
||||||
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) ───────────────────
|
||||||
@@ -188,12 +153,7 @@ with DAG(
|
|||||||
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
|
bash_command=f"cd {PIPELINE_DIR} && python scripts/sync_typesense.py",
|
||||||
)
|
)
|
||||||
|
|
||||||
invalidate_cache_ees = BashOperator(
|
extract_ees_group >> dbt_build_ees >> sync_typesense_ees
|
||||||
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) ────────────────────────────────────
|
||||||
@@ -218,9 +178,4 @@ 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+",
|
||||||
)
|
)
|
||||||
|
|
||||||
invalidate_cache_idaci = BashOperator(
|
extract_idaci >> dbt_build_idaci
|
||||||
task_id="invalidate_cache",
|
|
||||||
bash_command=INVALIDATE_CACHE_CMD,
|
|
||||||
)
|
|
||||||
|
|
||||||
extract_idaci >> dbt_build_idaci >> invalidate_cache_idaci
|
|
||||||
|
|||||||
@@ -94,23 +94,13 @@ def main() -> None:
|
|||||||
for code_col, name_col, dict_name, field_key in FIELDS:
|
for code_col, name_col, dict_name, field_key in FIELDS:
|
||||||
pairs = (
|
pairs = (
|
||||||
df[[code_col, name_col]]
|
df[[code_col, name_col]]
|
||||||
.loc[lambda d: d[code_col] != ""]
|
.loc[lambda d: (d[code_col] != "") & (d[name_col] != "")]
|
||||||
.drop_duplicates()
|
.drop_duplicates()
|
||||||
)
|
)
|
||||||
by_code: dict[int, set] = {}
|
mapping = sorted((int(c), n) for c, n in pairs.itertuples(index=False))
|
||||||
for c, n in pairs.itertuples(index=False):
|
dupes = len(mapping) - len({c for c, _ in mapping})
|
||||||
by_code.setdefault(int(c), set()).add(n)
|
if dupes:
|
||||||
mapping = []
|
sys.exit(f"{code_col}: {dupes} codes map to multiple names — investigate before generating")
|
||||||
for code, names in sorted(by_code.items()):
|
|
||||||
named = sorted(n for n in names if n != "")
|
|
||||||
if len(named) > 1:
|
|
||||||
sys.exit(f"{code_col}: code {code} maps to multiple names {named} — investigate before generating")
|
|
||||||
# Codes that only ever appear with a blank (name) are GIAS
|
|
||||||
# "not recorded" sentinels (e.g. ReligiousCharacter 99,
|
|
||||||
# AdmissionsPolicy 9). Map them to "" so the API serves the same
|
|
||||||
# empty string the old name pipeline did — the "Unknown (<code>)"
|
|
||||||
# path is reserved for genuinely new codes.
|
|
||||||
mapping.append((code, named[0] if named else ""))
|
|
||||||
lines = [f"{dict_name}: dict[int, str] = {{"]
|
lines = [f"{dict_name}: dict[int, str] = {{"]
|
||||||
for code, name in mapping:
|
for code, name in mapping:
|
||||||
escaped = name.replace('"', '\\"')
|
escaped = name.replace('"', '\\"')
|
||||||
|
|||||||
@@ -78,7 +78,6 @@ OFFICIAL_SIXTH_FORM: dict[int, str] = {
|
|||||||
0: "Not applicable",
|
0: "Not applicable",
|
||||||
1: "Has a sixth form",
|
1: "Has a sixth form",
|
||||||
2: "Does not have a sixth form",
|
2: "Does not have a sixth form",
|
||||||
9: "",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
RELIGIOUS_CHARACTER: dict[int, str] = {
|
RELIGIOUS_CHARACTER: dict[int, str] = {
|
||||||
@@ -129,14 +128,12 @@ RELIGIOUS_CHARACTER: dict[int, str] = {
|
|||||||
47: "Reformed Baptist",
|
47: "Reformed Baptist",
|
||||||
48: "Roman Catholic/Anglican",
|
48: "Roman Catholic/Anglican",
|
||||||
49: "Sunni Deobandi",
|
49: "Sunni Deobandi",
|
||||||
99: "",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
ADMISSIONS_POLICY: dict[int, str] = {
|
ADMISSIONS_POLICY: dict[int, str] = {
|
||||||
0: "Not applicable",
|
0: "Not applicable",
|
||||||
2: "Selective",
|
2: "Selective",
|
||||||
4: "Non-selective",
|
4: "Non-selective",
|
||||||
9: "",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -42,12 +42,12 @@ models:
|
|||||||
tests:
|
tests:
|
||||||
- accepted_values:
|
- accepted_values:
|
||||||
severity: warn
|
severity: warn
|
||||||
values: [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 24, 25, 26, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 99]
|
values: [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 24, 25, 26, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49]
|
||||||
- name: admissions_policy_code
|
- name: admissions_policy_code
|
||||||
tests:
|
tests:
|
||||||
- accepted_values:
|
- accepted_values:
|
||||||
severity: warn
|
severity: warn
|
||||||
values: [0, 2, 4, 9]
|
values: [0, 2, 4]
|
||||||
|
|
||||||
- name: dim_location
|
- name: dim_location
|
||||||
description: School location dimension with PostGIS geometry
|
description: School location dimension with PostGIS geometry
|
||||||
|
|||||||
@@ -53,7 +53,6 @@ phase_of_education,7,All-through
|
|||||||
official_sixth_form,0,Not applicable
|
official_sixth_form,0,Not applicable
|
||||||
official_sixth_form,1,Has a sixth form
|
official_sixth_form,1,Has a sixth form
|
||||||
official_sixth_form,2,Does not have a sixth form
|
official_sixth_form,2,Does not have a sixth form
|
||||||
official_sixth_form,9,
|
|
||||||
religious_character,0,Does not apply
|
religious_character,0,Does not apply
|
||||||
religious_character,2,Church of England
|
religious_character,2,Church of England
|
||||||
religious_character,3,Roman Catholic
|
religious_character,3,Roman Catholic
|
||||||
@@ -101,8 +100,6 @@ religious_character,46,Protestant/Evangelical
|
|||||||
religious_character,47,Reformed Baptist
|
religious_character,47,Reformed Baptist
|
||||||
religious_character,48,Roman Catholic/Anglican
|
religious_character,48,Roman Catholic/Anglican
|
||||||
religious_character,49,Sunni Deobandi
|
religious_character,49,Sunni Deobandi
|
||||||
religious_character,99,
|
|
||||||
admissions_policy,0,Not applicable
|
admissions_policy,0,Not applicable
|
||||||
admissions_policy,2,Selective
|
admissions_policy,2,Selective
|
||||||
admissions_policy,4,Non-selective
|
admissions_policy,4,Non-selective
|
||||||
admissions_policy,9,
|
|
||||||
|
|||||||
|
Reference in New Issue
Block a user