TudorandClaude Opus 5
2e9b5c83c5
PR Checks / Frontend Typecheck + Tests (pull_request) Successful in 1m5s
PR Checks / Backend Smoke (pull_request) Successful in 9s
PR Checks / Build Backend (no push) (pull_request) Successful in 18s
PR Checks / Build Frontend (no push) (pull_request) Successful in 45s
PR Checks / Build Pipeline (no push) (pull_request) Successful in 1m16s
PR Checks / AI Code Review (Claude) (pull_request) Successful in 7m23s
Review found _mask_for_disclosure could return with its invariant broken and say nothing. add_companion only ever withheld a *published* cell, so a group with one suppressed category and every other one not_applicable — routine in special schools and AP, where few categories apply — left the loop with the lone suppressed cell still solvable. Reproduced on a nine-pupil cohort: one hidden cell, cohort served, residual intact. A disclosure-control pass that fails silently is worse than none, because everything downstream trusts it. The loop now runs until the invariant holds and escalates when no companion exists: the pupil group is dropped from the payload, and an empty block serialises as None so the section is absent rather than an empty shell. disclosure_invariant_holds() is exported so tests assert it directly instead of re-deriving it, and an exhaustive test sweeps all 81 suppression patterns of a four-category group. Also fixes a test that set up six measures and checked one: the loop was `for measure in ["school_sixth_form"]`. It now checks every measure, and against the real invariant — none hidden, or at least two, rather than "at least two", which the five published measures would have failed. No regression on real data: 262 mainstream secondaries, all-pupils bar still drawable on 94%, zero invariant violations, one disadvantaged group dropped by the new escalation. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BvdDKvFFSZuMVDH5fEyTob
1209 lines
43 KiB
Python
1209 lines
43 KiB
Python
"""
|
|
Data loading module — reads from marts.* tables built by dbt.
|
|
Provides efficient queries with caching.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from typing import Optional, Dict, Tuple, List
|
|
import requests
|
|
from sqlalchemy import text
|
|
import sqlalchemy.exc
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .database import SessionLocal, engine
|
|
from .models import (
|
|
DimSchool, DimLocation, KS2Performance,
|
|
FactOfstedInspection, FactAdmissions, FactAdmissionDistance,
|
|
FactDeprivation, FactFinance, FactPupilCharacteristics,
|
|
FactKs4Destinations, FactKs5Destinations,
|
|
)
|
|
from .ofsted_codes import ofsted_page_url, report_card_labels
|
|
from .schemas import SCHOOL_TYPE_MAP
|
|
from .gias_codes import (
|
|
ADMISSIONS_POLICY,
|
|
ESTABLISHMENT_STATUS,
|
|
PHASE_OF_EDUCATION,
|
|
RELIGIOUS_CHARACTER,
|
|
SCHOOL_TYPE,
|
|
translate,
|
|
)
|
|
|
|
# mart code column -> (API name column, dictionary)
|
|
_GIAS_CODE_COLUMNS = {
|
|
"phase_code": ("phase", PHASE_OF_EDUCATION),
|
|
"school_type_code": ("school_type", SCHOOL_TYPE),
|
|
"status_code": ("status", ESTABLISHMENT_STATUS),
|
|
"religious_character_code": ("religious_denomination", RELIGIOUS_CHARACTER),
|
|
"admissions_policy_code": ("admissions_policy", ADMISSIONS_POLICY),
|
|
}
|
|
|
|
|
|
def translate_gias_code_columns(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Map GIAS code columns to today's name columns (API contract).
|
|
|
|
Runs immediately after pd.read_sql so every downstream consumer —
|
|
filters, PHASE_GROUPS, payloads, /api/filters — keeps seeing names.
|
|
DataFrames without the code columns (old schema, test fixtures) pass
|
|
through unchanged.
|
|
"""
|
|
for code_col, (name_col, mapping) in _GIAS_CODE_COLUMNS.items():
|
|
if code_col in df.columns:
|
|
df[name_col] = df[code_col].map(lambda c: translate(c, mapping))
|
|
return df
|
|
|
|
|
|
_postcode_cache: Dict[str, Tuple[float, float]] = {}
|
|
_typesense_client = None
|
|
|
|
|
|
def _get_typesense_client():
|
|
global _typesense_client
|
|
if _typesense_client is not None:
|
|
return _typesense_client
|
|
url = settings.typesense_url
|
|
key = settings.typesense_api_key
|
|
if not url or not key:
|
|
return None
|
|
try:
|
|
import typesense
|
|
host = url.split("//")[-1]
|
|
host_part, _, port_str = host.partition(":")
|
|
port = int(port_str) if port_str else 8108
|
|
_typesense_client = typesense.Client({
|
|
"nodes": [{"host": host_part, "port": str(port), "protocol": "http"}],
|
|
"api_key": key,
|
|
"connection_timeout_seconds": 2,
|
|
})
|
|
return _typesense_client
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def search_schools_typesense(query: str, limit: int = 250) -> List[int]:
|
|
"""Search Typesense. Returns URNs in relevance order, or [] if unavailable."""
|
|
client = _get_typesense_client()
|
|
if client is None:
|
|
return []
|
|
try:
|
|
result = client.collections["schools"].documents.search({
|
|
"q": query,
|
|
"query_by": "school_name,local_authority,postcode",
|
|
"per_page": min(limit, 250),
|
|
"typo_tokens_threshold": 1,
|
|
})
|
|
return [int(h["document"]["urn"]) for h in result.get("hits", [])]
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
# The most a public endpoint will return in one response.
|
|
SUGGEST_MAX_LIMIT = 20
|
|
|
|
# Fields a suggestion row carries, and the default when the document omits an
|
|
# optional one. phase and school_type are optional in the Typesense schema.
|
|
_SUGGEST_FIELDS = ("school_name", "local_authority", "postcode",
|
|
"phase", "school_type")
|
|
|
|
|
|
def suggest_schools_typesense(query: str, limit: int = 8) -> List[dict]:
|
|
"""Autosuggest rows straight from Typesense. Never raises.
|
|
|
|
Returns documents rather than URNs, unlike search_schools_typesense, so the
|
|
caller needs no DataFrame. Every field below is already in the index — see
|
|
pipeline/scripts/sync_typesense.py — which is what makes this cheap enough
|
|
to run per keystroke.
|
|
"""
|
|
client = _get_typesense_client()
|
|
if client is None:
|
|
return []
|
|
try:
|
|
result = client.collections["schools"].documents.search({
|
|
"q": query,
|
|
"query_by": "school_name,local_authority",
|
|
"per_page": max(1, min(limit, SUGGEST_MAX_LIMIT)),
|
|
"typo_tokens_threshold": 1,
|
|
})
|
|
except Exception:
|
|
# A dropdown that quietly stops appearing is the right failure here.
|
|
return []
|
|
|
|
rows = []
|
|
for hit in result.get("hits", []) or []:
|
|
doc = (hit or {}).get("document") or {}
|
|
try:
|
|
urn = int(doc["urn"])
|
|
except (KeyError, TypeError, ValueError):
|
|
# Skip the row, keep the rest. Typesense declares urn as int32 so
|
|
# this should be unreachable, but the index is a separate system
|
|
# that something other than this code can reindex — and "never
|
|
# raises" is a promise the keystroke path actually depends on.
|
|
# Dropping one malformed document is right; blanking the whole
|
|
# dropdown, or serving a suggestion pointing at /school/0, is not.
|
|
logging.getLogger(__name__).warning(
|
|
"skipping malformed suggestion document: %r", doc)
|
|
continue
|
|
row = {"urn": urn}
|
|
row.update({f: str(doc.get(f, "") or "") for f in _SUGGEST_FIELDS})
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def normalize_school_type(school_type: Optional[str]) -> Optional[str]:
|
|
"""Convert cryptic school type codes to user-friendly names."""
|
|
if not school_type:
|
|
return None
|
|
code = school_type.strip().upper()
|
|
if code in SCHOOL_TYPE_MAP:
|
|
return SCHOOL_TYPE_MAP[code]
|
|
return school_type
|
|
|
|
|
|
def geocode_single_postcode(postcode: str) -> Optional[Tuple[float, float]]:
|
|
"""Geocode a single postcode using postcodes.io API."""
|
|
if not postcode:
|
|
return None
|
|
postcode = postcode.strip().upper()
|
|
if postcode in _postcode_cache:
|
|
return _postcode_cache[postcode]
|
|
try:
|
|
response = requests.get(
|
|
f"https://api.postcodes.io/postcodes/{postcode}",
|
|
timeout=10,
|
|
)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("result"):
|
|
lat = data["result"].get("latitude")
|
|
lon = data["result"].get("longitude")
|
|
if lat and lon:
|
|
_postcode_cache[postcode] = (lat, lon)
|
|
return (lat, lon)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
|
|
"""Calculate great-circle distance between two points (miles)."""
|
|
from math import radians, cos, sin, asin, sqrt
|
|
lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
|
|
dlat = lat2 - lat1
|
|
dlon = lon2 - lon1
|
|
a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
|
|
return 2 * asin(sqrt(a)) * 3956
|
|
|
|
|
|
# =============================================================================
|
|
# MAIN DATA LOAD — joins dim_school + dim_location + fact_performance
|
|
# fact_performance is a merged KS2+KS4 table (one row per URN per year).
|
|
# All-through schools have both KS2 and KS4 columns populated in the same row.
|
|
# =============================================================================
|
|
|
|
_MAIN_QUERY = text("""
|
|
SELECT
|
|
s.urn,
|
|
s.school_name,
|
|
s.phase_code,
|
|
s.school_type_code,
|
|
s.academy_trust_name AS trust_name,
|
|
s.academy_trust_uid AS trust_uid,
|
|
s.religious_character_code,
|
|
s.gender,
|
|
s.age_range,
|
|
s.has_sixth_form,
|
|
s.status_code,
|
|
s.admissions_policy_code,
|
|
s.capacity,
|
|
s.total_pupils AS gias_total_pupils,
|
|
s.headteacher_name,
|
|
s.website,
|
|
s.telephone,
|
|
s.nursery_provision,
|
|
foi.ofsted_grade,
|
|
foi.ofsted_date,
|
|
foi.ofsted_framework,
|
|
foi.ofsted_rc_date,
|
|
l.local_authority_name AS local_authority,
|
|
l.local_authority_code,
|
|
l.address_line1 AS address1,
|
|
l.address_line2 AS address2,
|
|
l.town,
|
|
l.county,
|
|
l.parliamentary_constituency,
|
|
l.postcode,
|
|
l.latitude,
|
|
l.longitude,
|
|
p.year,
|
|
p.source_urn,
|
|
p.total_pupils,
|
|
p.eligible_pupils,
|
|
-- KS2 columns (NULL for pure secondary schools)
|
|
p.rwm_expected_pct,
|
|
p.rwm_high_pct,
|
|
p.reading_expected_pct,
|
|
p.reading_high_pct,
|
|
p.reading_avg_score,
|
|
p.reading_progress,
|
|
p.reading_progress_lower_ci,
|
|
p.reading_progress_upper_ci,
|
|
p.writing_expected_pct,
|
|
p.writing_high_pct,
|
|
p.writing_progress,
|
|
p.writing_progress_lower_ci,
|
|
p.writing_progress_upper_ci,
|
|
p.writing_working_towards_pct,
|
|
p.maths_expected_pct,
|
|
p.maths_high_pct,
|
|
p.maths_avg_score,
|
|
p.maths_progress,
|
|
p.maths_progress_lower_ci,
|
|
p.maths_progress_upper_ci,
|
|
p.gps_expected_pct,
|
|
p.gps_high_pct,
|
|
p.gps_avg_score,
|
|
p.science_expected_pct,
|
|
p.reading_absence_pct,
|
|
p.writing_absence_pct,
|
|
p.maths_absence_pct,
|
|
p.gps_absence_pct,
|
|
p.science_absence_pct,
|
|
p.rwm_expected_boys_pct,
|
|
p.rwm_high_boys_pct,
|
|
p.rwm_expected_girls_pct,
|
|
p.rwm_high_girls_pct,
|
|
p.rwm_expected_disadvantaged_pct,
|
|
p.rwm_expected_non_disadvantaged_pct,
|
|
p.disadvantaged_gap,
|
|
p.disadvantaged_pct,
|
|
p.eal_pct,
|
|
p.stability_pct,
|
|
-- KS4 columns (NULL for pure primary schools)
|
|
p.attainment_8_score,
|
|
p.progress_8_score,
|
|
p.progress_8_lower_ci,
|
|
p.progress_8_upper_ci,
|
|
p.progress_8_english,
|
|
p.progress_8_maths,
|
|
p.progress_8_ebacc,
|
|
p.progress_8_open,
|
|
p.progress_8_banding,
|
|
p.attainment_8_disadvantage_gap,
|
|
p.progress_8_disadvantage_gap,
|
|
p.english_maths_strong_pass_pct,
|
|
p.english_maths_standard_pass_pct,
|
|
p.ebacc_entry_pct,
|
|
p.ebacc_strong_pass_pct,
|
|
p.ebacc_standard_pass_pct,
|
|
p.ebacc_avg_score,
|
|
p.gcse_grade_91_pct,
|
|
p.prior_attainment_avg,
|
|
-- SEN (coalesced KS2+KS4 in fact_performance)
|
|
p.sen_support_pct,
|
|
p.sen_ehcp_pct
|
|
FROM marts.dim_school s
|
|
JOIN marts.dim_location l ON s.urn = l.urn
|
|
LEFT JOIN marts.fact_performance p ON s.urn = p.urn
|
|
LEFT JOIN (
|
|
SELECT DISTINCT ON (urn)
|
|
urn,
|
|
-- Fall back to the ungraded-inspection grade when no graded grade exists.
|
|
COALESCE(overall_effectiveness, ungraded_grade) AS ofsted_grade,
|
|
inspection_date AS ofsted_date,
|
|
framework AS ofsted_framework,
|
|
-- Report-card signal for list/map badges: non-null only when the
|
|
-- latest inspection carries report-card grades. framework is the
|
|
-- raw event grouping ("Schools - S5"), never "ReportCard", so it
|
|
-- can't be used to detect report cards.
|
|
rc_inspection_date AS ofsted_rc_date
|
|
FROM marts.fact_ofsted_inspection
|
|
ORDER BY urn, inspection_date DESC NULLS LAST
|
|
) foi ON s.urn = foi.urn
|
|
ORDER BY s.school_name, p.year
|
|
""")
|
|
|
|
# Fallback used when marts.dim_school predates the has_sixth_form column
|
|
# (i.e. the nightly dbt pipeline hasn't rebuilt the mart yet on this DB).
|
|
# Keeps the column present as NULL so downstream code — including the
|
|
# app.py fallback branch — behaves as designed instead of KeyError-ing.
|
|
_MAIN_QUERY_NO_SIXTH_FORM = text(
|
|
str(_MAIN_QUERY).replace("s.has_sixth_form,", "NULL AS has_sixth_form,")
|
|
)
|
|
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)
|
|
|
|
# Fallback used when marts predate the optional GIAS/location detail columns
|
|
# (telephone, nursery provision, county, parliamentary constituency) — i.e. the
|
|
# nightly dbt pipeline hasn't rebuilt the mart yet on this DB. Keeps each column
|
|
# present as NULL so the whole data load degrades gracefully instead of failing.
|
|
_EXTRA_OPTIONAL_COLUMNS = (
|
|
"telephone",
|
|
"nursery_provision",
|
|
"county",
|
|
"parliamentary_constituency",
|
|
)
|
|
_MAIN_QUERY_NO_EXTRA_COLS = str(_MAIN_QUERY)
|
|
for _col, _tbl in (
|
|
("telephone", "s"),
|
|
("nursery_provision", "s"),
|
|
("county", "l"),
|
|
("parliamentary_constituency", "l"),
|
|
):
|
|
_src = f"{_tbl}.{_col},"
|
|
assert _src in _MAIN_QUERY_NO_EXTRA_COLS, (
|
|
f"expected {_src!r} to be present in _MAIN_QUERY before replacement"
|
|
)
|
|
_MAIN_QUERY_NO_EXTRA_COLS = _MAIN_QUERY_NO_EXTRA_COLS.replace(
|
|
_src, f"NULL AS {_col},"
|
|
)
|
|
_MAIN_QUERY_NO_EXTRA_COLS = text(_MAIN_QUERY_NO_EXTRA_COLS)
|
|
|
|
_GIAS_CODE_COLUMN_NAMES = (
|
|
"phase_code",
|
|
"school_type_code",
|
|
"religious_character_code",
|
|
"status_code",
|
|
"admissions_policy_code",
|
|
)
|
|
|
|
# Strip any single table alias prefix (s., l., p., foi., …) — Postgres reports
|
|
# an undefined *qualified* column unquoted as "column l.county does not exist",
|
|
# so matching only the s. alias would miss dim_location columns (county,
|
|
# parliamentary_constituency) and defeat the graceful-degradation fallback.
|
|
_MISSING_COLUMN_RE = re.compile(r'column "?(?:\w+\.)?(\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:
|
|
"""Load all school + KS2 data as a pandas DataFrame."""
|
|
try:
|
|
df = pd.read_sql(_MAIN_QUERY, engine)
|
|
except sqlalchemy.exc.ProgrammingError as exc:
|
|
missing = _missing_column_name(exc)
|
|
if missing in _GIAS_CODE_COLUMN_NAMES:
|
|
logging.getLogger(__name__).warning(
|
|
"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 missing == "has_sixth_form":
|
|
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",
|
|
exc,
|
|
)
|
|
try:
|
|
df = pd.read_sql(_MAIN_QUERY_NO_SIXTH_FORM, engine)
|
|
except Exception as exc2:
|
|
print(f"Warning: Could not load school data from marts: {exc2}")
|
|
return pd.DataFrame()
|
|
elif missing in _EXTRA_OPTIONAL_COLUMNS:
|
|
logging.getLogger(__name__).warning(
|
|
"marts.dim_school/dim_location is missing %s (pipeline hasn't "
|
|
"rebuilt the mart yet on this DB) — retrying without the "
|
|
"optional detail columns: %s",
|
|
missing, exc,
|
|
)
|
|
try:
|
|
df = pd.read_sql(_MAIN_QUERY_NO_EXTRA_COLS, engine)
|
|
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()
|
|
|
|
if df.empty:
|
|
return df
|
|
|
|
df = translate_gias_code_columns(df)
|
|
|
|
# Build address string
|
|
df["address"] = df.apply(
|
|
lambda r: ", ".join(
|
|
p for p in [r.get("address1"), r.get("address2"), r.get("town"), r.get("postcode")]
|
|
if p and str(p) != "None"
|
|
),
|
|
axis=1,
|
|
)
|
|
|
|
# Normalize school type
|
|
df["school_type"] = df["school_type"].apply(normalize_school_type)
|
|
|
|
return df
|
|
|
|
|
|
# Cache for DataFrame
|
|
_df_cache: Optional[pd.DataFrame] = None
|
|
# Pre-computed latest-year snapshot (one row per school, with prev-year trend columns)
|
|
_df_latest_cache: Optional[pd.DataFrame] = None
|
|
|
|
|
|
def load_school_data() -> pd.DataFrame:
|
|
"""Load school data with caching."""
|
|
global _df_cache
|
|
if _df_cache is not None:
|
|
return _df_cache
|
|
print("Loading school data from marts...")
|
|
_df_cache = load_school_data_as_dataframe()
|
|
if not _df_cache.empty:
|
|
print(f"Total records loaded: {len(_df_cache)}")
|
|
print(f"Unique schools: {_df_cache['urn'].nunique()}")
|
|
print(f"Years: {sorted(_df_cache['year'].dropna().unique())}")
|
|
else:
|
|
print("No data found in marts (EES data may not have been loaded yet)")
|
|
return _df_cache
|
|
|
|
|
|
def load_latest_school_data() -> pd.DataFrame:
|
|
"""Return a cached one-row-per-school DataFrame at the latest available year.
|
|
|
|
The expensive groupby / merge / prev-year trend computation runs once at
|
|
startup (or after a cache clear) rather than on every search request.
|
|
Per-request filters (phase, gender, LA …) should be applied to the returned
|
|
DataFrame's copy; they must NOT modify the cached object.
|
|
"""
|
|
global _df_latest_cache
|
|
if _df_latest_cache is not None:
|
|
return _df_latest_cache
|
|
|
|
df = load_school_data()
|
|
if df.empty:
|
|
return df
|
|
|
|
# Schools that have no performance rows (PRUs, new schools, etc.)
|
|
df_no_perf = df[df["year"].isna()].drop_duplicates(subset=["urn"])
|
|
df_with_perf = df[df["year"].notna()]
|
|
|
|
# Reduce to the latest year per school
|
|
latest_year = df_with_perf.groupby("urn")["year"].max().reset_index()
|
|
df_latest = df_with_perf.merge(latest_year, on=["urn", "year"])
|
|
|
|
# Attach previous-year metrics for trend arrows (second-latest year per school)
|
|
df_sorted = df_with_perf.sort_values(["urn", "year"], ascending=[True, False])
|
|
df_prev = df_sorted.groupby("urn").nth(1).reset_index()
|
|
if not df_prev.empty and "rwm_expected_pct" in df_prev.columns:
|
|
prev_rwm = df_prev[["urn", "rwm_expected_pct"]].rename(
|
|
columns={"rwm_expected_pct": "prev_rwm_expected_pct"}
|
|
)
|
|
if "attainment_8_score" in df_prev.columns:
|
|
prev_rwm = prev_rwm.merge(
|
|
df_prev[["urn", "attainment_8_score"]].rename(
|
|
columns={"attainment_8_score": "prev_attainment_8_score"}
|
|
),
|
|
on="urn",
|
|
how="outer",
|
|
)
|
|
df_latest = df_latest.merge(prev_rwm, on="urn", how="left")
|
|
|
|
# Merge back schools with no performance data
|
|
df_latest = pd.concat([df_latest, df_no_perf], ignore_index=True)
|
|
|
|
print(f"Latest-snapshot cache built: {len(df_latest)} schools")
|
|
_df_latest_cache = df_latest
|
|
return _df_latest_cache
|
|
|
|
|
|
def clear_cache():
|
|
"""Clear all caches."""
|
|
global _df_cache, _df_latest_cache
|
|
_df_cache = None
|
|
_df_latest_cache = None
|
|
|
|
|
|
# =============================================================================
|
|
# METADATA QUERIES
|
|
# =============================================================================
|
|
|
|
def get_available_years(db: Session = None) -> List[int]:
|
|
close_db = db is None
|
|
if db is None:
|
|
db = SessionLocal()
|
|
try:
|
|
result = db.query(KS2Performance.year).distinct().order_by(KS2Performance.year).all()
|
|
return [r[0] for r in result]
|
|
except Exception:
|
|
return []
|
|
finally:
|
|
if close_db:
|
|
db.close()
|
|
|
|
|
|
def get_available_local_authorities(db: Session = None) -> List[str]:
|
|
close_db = db is None
|
|
if db is None:
|
|
db = SessionLocal()
|
|
try:
|
|
result = (
|
|
db.query(DimLocation.local_authority_name)
|
|
.filter(DimLocation.local_authority_name.isnot(None))
|
|
.distinct()
|
|
.order_by(DimLocation.local_authority_name)
|
|
.all()
|
|
)
|
|
return [r[0] for r in result if r[0]]
|
|
except Exception:
|
|
return []
|
|
finally:
|
|
if close_db:
|
|
db.close()
|
|
|
|
|
|
def get_schools_count(db: Session = None) -> int:
|
|
close_db = db is None
|
|
if db is None:
|
|
db = SessionLocal()
|
|
try:
|
|
return db.query(DimSchool).count()
|
|
except Exception:
|
|
return 0
|
|
finally:
|
|
if close_db:
|
|
db.close()
|
|
|
|
|
|
def get_data_info(db: Session = None) -> dict:
|
|
close_db = db is None
|
|
if db is None:
|
|
db = SessionLocal()
|
|
try:
|
|
school_count = get_schools_count(db)
|
|
years = get_available_years(db)
|
|
local_authorities = get_available_local_authorities(db)
|
|
return {
|
|
"total_schools": school_count,
|
|
"years_available": years,
|
|
"local_authorities_count": len(local_authorities),
|
|
"data_source": "PostgreSQL (marts)",
|
|
}
|
|
finally:
|
|
if close_db:
|
|
db.close()
|
|
|
|
|
|
# =============================================================================
|
|
# SUPPLEMENTARY DATA — per-school detail page
|
|
# =============================================================================
|
|
|
|
def compute_benchmarks(df: pd.DataFrame, census_benchmarks: dict | None = None) -> dict:
|
|
"""State-school benchmarks computed from our dataset (spec §5/§8.6).
|
|
|
|
NOT official DfE figures — consumers must label them
|
|
"state-school average (computed from our dataset)". The disadvantaged
|
|
attainment average is weighted by cohort size (eligible_pupils) so
|
|
small schools don't dominate.
|
|
|
|
Context measures (FSM/EAL/pupil counts) come from `census_benchmarks`
|
|
(the fact_census_benchmarks mart, pupil-weighted, keyed by phase): the
|
|
performance df has no fsm_pct at all, and its eal/disadvantaged columns
|
|
are KS2-only — medianing them for "secondary" produced junk anchors
|
|
from the handful of all-through schools. When the mart is unavailable
|
|
these are None; never fall back across measure definitions.
|
|
"""
|
|
if df.empty or "year" not in df.columns:
|
|
return {}
|
|
latest_year = df["year"].max()
|
|
if pd.isna(latest_year):
|
|
return {}
|
|
d = df[df["year"] == latest_year]
|
|
if d.empty:
|
|
return {}
|
|
is_secondary = (
|
|
d["attainment_8_score"].notna()
|
|
if "attainment_8_score" in d.columns
|
|
else pd.Series(False, index=d.index)
|
|
)
|
|
prim, sec = d[~is_secondary], d[is_secondary]
|
|
|
|
def _median(sub, col):
|
|
if col not in sub.columns:
|
|
return None
|
|
v = sub[col].median()
|
|
return round(float(v), 1) if pd.notna(v) else None
|
|
|
|
def _weighted_disadvantaged(sub):
|
|
needed = {"rwm_expected_disadvantaged_pct", "eligible_pupils"}
|
|
if not needed <= set(sub.columns):
|
|
return None
|
|
s = sub.dropna(subset=list(needed))
|
|
if s.empty or s["eligible_pupils"].sum() == 0:
|
|
return None
|
|
w = (
|
|
(s["rwm_expected_disadvantaged_pct"] * s["eligible_pupils"]).sum()
|
|
/ s["eligible_pupils"].sum()
|
|
)
|
|
return round(float(w), 1)
|
|
|
|
def _block(sub, phase, with_disadvantaged):
|
|
census = (census_benchmarks or {}).get(phase) or {}
|
|
block = {
|
|
"eal_pct": census.get("eal_pct"),
|
|
"sen_support_pct": _median(sub, "sen_support_pct"),
|
|
"disadvantaged_pct": _median(sub, "disadvantaged_pct") if with_disadvantaged else None,
|
|
"fsm_pct": census.get("fsm_pct"),
|
|
"median_pupils": census.get("median_pupils"),
|
|
}
|
|
if with_disadvantaged:
|
|
block["disadvantaged_rwm_expected_pct"] = _weighted_disadvantaged(sub)
|
|
return block
|
|
|
|
return {
|
|
"source": "state-school average (computed from our dataset)",
|
|
"year": int(latest_year),
|
|
"primary": _block(prim, "primary", with_disadvantaged=True),
|
|
"secondary": _block(sec, "secondary", with_disadvantaged=False),
|
|
}
|
|
|
|
|
|
def _ofsted_block(o, urn: int) -> dict:
|
|
"""Serialize the latest Ofsted inspection row for API responses.
|
|
|
|
`grade_source` records where the effective overall grade came from:
|
|
a graded (Section 5) inspection, or carried forward from an ungraded
|
|
(Section 8) outcome — materially different claims a UI must be able
|
|
to distinguish. `report_card` holds coded+labelled renewed-framework
|
|
(Nov 2025) area judgements; safeguarding is a separate boolean and
|
|
never appears among the graded areas.
|
|
"""
|
|
if o.overall_effectiveness is not None:
|
|
grade_source = "graded"
|
|
overall = o.overall_effectiveness
|
|
elif o.ungraded_grade is not None:
|
|
# Fall back to the grade parsed from an ungraded (Section 8) outcome
|
|
# (e.g. "School remains Good") so the detail page matches the list badge.
|
|
grade_source = "ungraded_carried_forward"
|
|
overall = o.ungraded_grade
|
|
else:
|
|
grade_source = None
|
|
overall = None
|
|
|
|
block = {
|
|
"framework": o.framework,
|
|
"inspection_date": o.inspection_date.isoformat() if o.inspection_date else None,
|
|
"rc_inspection_date": (
|
|
o.rc_inspection_date.isoformat()
|
|
if getattr(o, "rc_inspection_date", None)
|
|
else None
|
|
),
|
|
"inspection_type": o.inspection_type,
|
|
"overall_effectiveness": overall,
|
|
"grade_source": grade_source,
|
|
"quality_of_education": o.quality_of_education,
|
|
"behaviour_attitudes": o.behaviour_attitudes,
|
|
"personal_development": o.personal_development,
|
|
"leadership_management": o.leadership_management,
|
|
"early_years_provision": o.early_years_provision,
|
|
"sixth_form_provision": o.sixth_form_provision,
|
|
"previous_overall": None, # Not available in new schema
|
|
"rc_safeguarding_met": o.rc_safeguarding_met,
|
|
"rc_inclusion": o.rc_inclusion,
|
|
"rc_curriculum_teaching": o.rc_curriculum_teaching,
|
|
"rc_achievement": o.rc_achievement,
|
|
"rc_attendance_behaviour": o.rc_attendance_behaviour,
|
|
"rc_personal_development": o.rc_personal_development,
|
|
"rc_leadership_governance": o.rc_leadership_governance,
|
|
"rc_early_years": o.rc_early_years,
|
|
"rc_sixth_form": o.rc_sixth_form,
|
|
"report_url": o.report_url,
|
|
"ofsted_page_url": ofsted_page_url(urn),
|
|
}
|
|
block["report_card"] = report_card_labels(block)
|
|
return block
|
|
|
|
|
|
def _admissions_row_dict(a) -> dict:
|
|
"""Serialize one fact_admissions row for API responses."""
|
|
return {
|
|
"year": a.year,
|
|
"school_phase": a.school_phase,
|
|
"places_offered": a.places_offered,
|
|
"total_applications": a.total_applications,
|
|
"first_preference_applications": a.first_preference_applications,
|
|
"first_preference_offers": a.first_preference_offers,
|
|
"first_preference_offer_pct": a.first_preference_offer_pct,
|
|
"oversubscription_ratio": a.oversubscription_ratio,
|
|
"oversubscribed": a.oversubscribed,
|
|
"total_offers": a.total_offers,
|
|
"second_preference_offers": a.second_preference_offers,
|
|
"third_preference_offers": a.third_preference_offers,
|
|
"cross_la_applications": a.cross_la_applications,
|
|
"cross_la_offers": a.cross_la_offers,
|
|
}
|
|
|
|
|
|
def _admission_distance_dict(d) -> dict:
|
|
"""Serialize one fact_admission_distance row for API responses."""
|
|
return {
|
|
"year": d.year,
|
|
"distance_m": d.distance_m,
|
|
"route_count": d.route_count,
|
|
"la_name": d.la_name,
|
|
"distance_unit_raw": d.distance_unit_raw,
|
|
}
|
|
|
|
|
|
def _census_dict(pc) -> dict:
|
|
return {
|
|
"year": pc.year,
|
|
"total_pupils": pc.total_pupils,
|
|
"female_pupils": pc.female_pupils,
|
|
"male_pupils": pc.male_pupils,
|
|
"fsm_pct": pc.fsm_pct,
|
|
"eal_pct": pc.eal_pct,
|
|
}
|
|
|
|
|
|
def _deprivation_dict(d) -> dict:
|
|
return {
|
|
"lsoa_code": d.lsoa_code,
|
|
"idaci_score": d.idaci_score,
|
|
"idaci_decile": d.idaci_decile,
|
|
}
|
|
|
|
|
|
def _finance_dict(f) -> dict:
|
|
return {
|
|
"year": f.year,
|
|
"per_pupil_spend": f.per_pupil_spend,
|
|
"staff_cost_pct": f.staff_cost_pct,
|
|
"teacher_cost_pct": f.teacher_cost_pct,
|
|
"support_staff_cost_pct": f.support_staff_cost_pct,
|
|
"premises_cost_pct": f.premises_cost_pct,
|
|
}
|
|
|
|
|
|
# Destination measures that are totals DfE published itself, rather than one of
|
|
# the categories that partition the cohort.
|
|
_AGGREGATE_MEASURES = {"agg_sustained_education", "agg_sustained_all"}
|
|
|
|
|
|
def _format_cohort_year(year) -> str | None:
|
|
"""202223 -> '2022/23'.
|
|
|
|
The section has to date its own cohort. Destination measures run about two
|
|
GCSE years behind the results shown above them on the same page, so an
|
|
undated figure reads as stale data rather than as a different question.
|
|
"""
|
|
if not year:
|
|
return None
|
|
text = str(year)
|
|
if len(text) == 6:
|
|
return f"{text[:4]}/{text[4:6]}"
|
|
if len(text) == 8:
|
|
return f"{text[:4]}/{text[6:8]}"
|
|
return text
|
|
|
|
|
|
_PUPIL_GROUPS = ("disadvantaged", "other", "all")
|
|
|
|
|
|
def _lone_hidden_groups(groups: dict) -> list:
|
|
"""Pupil groups hiding exactly one category — solvable by subtraction."""
|
|
return [
|
|
key for key, group in groups.items()
|
|
if sum(1 for c in group["categories"] if c["status"] == "suppressed") == 1
|
|
]
|
|
|
|
|
|
def _lone_hidden_categories(groups: dict) -> list:
|
|
"""Categories hidden in exactly one of several pupil groups."""
|
|
lone = []
|
|
categories = {c["category"] for g in groups.values() for c in g["categories"]}
|
|
for category in categories:
|
|
found = [
|
|
c for g in groups.values() for c in g["categories"]
|
|
if c["category"] == category
|
|
]
|
|
hidden = [c for c in found if c["status"] == "suppressed"]
|
|
if len(hidden) == 1 and len(found) > 1:
|
|
lone.append(category)
|
|
return lone
|
|
|
|
|
|
def disclosure_invariant_holds(groups: dict) -> bool:
|
|
"""Every row and every column hides none, or at least two.
|
|
|
|
Public so the tests can assert it directly rather than re-deriving it.
|
|
"""
|
|
return not _lone_hidden_groups(groups) and not _lone_hidden_categories(groups)
|
|
|
|
|
|
def _mask_for_disclosure(groups: dict) -> None:
|
|
"""Withhold further cells until nothing suppressed can be solved for.
|
|
|
|
Not rendering a figure is not the same as not publishing it. This endpoint
|
|
is public and unauthenticated, so anything left in the payload is
|
|
published, whatever the UI chooses to draw — the same reasoning the
|
|
admission_distance field carries in app.py.
|
|
|
|
Two identities let a caller solve for a withheld cell:
|
|
|
|
* within a pupil group, the categories sum to the cohort, so a group with
|
|
exactly ONE suppressed category gives it away as cohort - sum(rest);
|
|
* across groups, disadvantaged + other = all for every category, so a
|
|
category suppressed in exactly ONE of the three gives itself away.
|
|
|
|
DfE's own answer is secondary suppression: withhold a second cell so the
|
|
residual spans two unknowns and identifies neither.
|
|
|
|
Where no companion can do that — a sparse cohort whose every other category
|
|
is `not_applicable`, which is common in special schools and alternative
|
|
provision — there is nothing left to withhold, so the pupil group is
|
|
DROPPED entirely. An earlier version simply gave up here and returned with
|
|
the violation intact and no signal, which is the one outcome this function
|
|
must never produce: a disclosure-control pass that fails silently is worse
|
|
than none, because everything downstream trusts it.
|
|
|
|
Mutates `groups` in place. Guaranteed to return with
|
|
disclosure_invariant_holds(groups) true.
|
|
"""
|
|
|
|
def suppress(cell):
|
|
if cell["status"] == "published":
|
|
cell["status"] = "suppressed"
|
|
cell["pupils"] = None
|
|
cell["percentage"] = None
|
|
return True
|
|
return False
|
|
|
|
def add_companion(candidates) -> bool:
|
|
"""Withhold a second cell so the residual spans two unknowns.
|
|
|
|
The companion must carry pupils. Suppressing a zero looks like
|
|
secondary suppression and protects nothing: the residual still equals
|
|
the original withheld figure exactly. Returns False when no cell can
|
|
do the job, which escalates to dropping the group.
|
|
"""
|
|
published = [c for c in candidates if c["status"] == "published"]
|
|
useful = sorted(
|
|
(c for c in published if (c["pupils"] or 0) > 0),
|
|
key=lambda c: c["pupils"],
|
|
)
|
|
if useful:
|
|
return suppress(useful[0])
|
|
# Every remaining cell is zero or not applicable: withholding any of
|
|
# them leaves the residual equal to the original figure.
|
|
return False
|
|
|
|
# Fixpoint: each new suppression can break the other identity. Terminates
|
|
# because every pass either adds a suppression, drops a group, or stops.
|
|
while not disclosure_invariant_holds(groups):
|
|
changed = False
|
|
|
|
for category in _lone_hidden_categories(groups):
|
|
siblings = [
|
|
c for g in groups.values() for c in g["categories"]
|
|
if c["category"] == category
|
|
]
|
|
if add_companion(siblings):
|
|
changed = True
|
|
|
|
for key in _lone_hidden_groups(groups):
|
|
if add_companion(groups[key]["categories"]):
|
|
changed = True
|
|
|
|
if changed:
|
|
continue
|
|
|
|
# Nothing left to withhold. Drop the groups that are still solvable,
|
|
# and any category still solvable across the groups that remain.
|
|
for key in _lone_hidden_groups(groups):
|
|
del groups[key]
|
|
changed = True
|
|
|
|
for category in _lone_hidden_categories(groups):
|
|
for group in groups.values():
|
|
for cell in group["categories"]:
|
|
if cell["category"] == category and suppress(cell):
|
|
changed = True
|
|
|
|
if not changed:
|
|
# Unreachable given the two escalations above, but a masking pass
|
|
# must never spin or exit unsafely. Withhold everything.
|
|
groups.clear()
|
|
return
|
|
|
|
|
|
def _destinations_block(rows: list) -> dict | None:
|
|
"""Shape destination rows for one phase into the API's block.
|
|
|
|
Applies secondary suppression before returning, so no caller of this public
|
|
endpoint can solve for a figure DfE withheld. See _mask_for_disclosure.
|
|
|
|
Aggregate measures are dropped entirely. DfE publishes them, and they would
|
|
be useful for a "what is published for this group" fallback, but nothing
|
|
renders them today and an aggregate spanning exactly one suppressed
|
|
component names that component. An unused field that leaks is not a
|
|
trade-off worth carrying — re-add them with their own guard if the fallback
|
|
is ever built.
|
|
|
|
Deliberately computes no residual, no "remaining pupils" figure, and no
|
|
total that would close a gap left by a suppressed category.
|
|
"""
|
|
if not rows:
|
|
return None
|
|
|
|
years = [r["year"] for r in rows if r.get("year") is not None]
|
|
if not years:
|
|
return None
|
|
latest_year = max(years)
|
|
rows = [r for r in rows if r.get("year") == latest_year]
|
|
|
|
groups: dict = {}
|
|
for row in rows:
|
|
group = groups.setdefault(
|
|
row["pupil_group"],
|
|
{"cohort": row.get("cohort_pupils"), "categories": []},
|
|
)
|
|
measure = row["destination_measure"]
|
|
published = row.get("status") == "published"
|
|
# Belt and braces: percentage is derived from the same source cell as
|
|
# pupils, but publishing one without the other would hand back the
|
|
# cohort (pupils / percentage) and with it the residual.
|
|
cell = {
|
|
"category": measure,
|
|
"pupils": row.get("pupils") if published else None,
|
|
"percentage": row.get("percentage") if published else None,
|
|
"status": row.get("status"),
|
|
}
|
|
if measure in _AGGREGATE_MEASURES:
|
|
continue
|
|
group["categories"].append(cell)
|
|
|
|
if not groups:
|
|
return None
|
|
|
|
_mask_for_disclosure(groups)
|
|
|
|
# Masking can empty the block entirely — a sparse cohort where no group
|
|
# could be made safe. Return None so the section is absent rather than
|
|
# rendering an empty shell.
|
|
if not groups:
|
|
return None
|
|
|
|
return {"cohort_year": _format_cohort_year(latest_year), "groups": groups}
|
|
|
|
|
|
def _empty_supplementary() -> dict:
|
|
return {
|
|
"ofsted": None,
|
|
"census": None,
|
|
"admissions": None,
|
|
"admissions_history": [],
|
|
"admission_distance": None,
|
|
"sen_detail": None,
|
|
"phonics": None,
|
|
"deprivation": None,
|
|
"finance": None,
|
|
"destinations": None,
|
|
}
|
|
|
|
|
|
def get_supplementary_data_batch(db: Session, urns: list[int]) -> dict:
|
|
"""Fetch supplementary data for many URNs with one query per table
|
|
(WHERE urn IN (...)) instead of ~5 queries per school, collapsing the
|
|
per-request round-trips from 5*N to a constant 5. Returns {urn: block}
|
|
with the same shape get_supplementary_data produces per URN.
|
|
|
|
Each table is queried independently and failures degrade that table to
|
|
empty for every URN — a missing mart never blanks the others.
|
|
"""
|
|
urns = [int(u) for u in urns]
|
|
result = {urn: _empty_supplementary() for urn in urns}
|
|
if not urns:
|
|
return result
|
|
|
|
def _safe(fn):
|
|
try:
|
|
fn()
|
|
except Exception as e:
|
|
import logging
|
|
logging.getLogger(__name__).error("batch supplementary query failed: %s", e)
|
|
db.rollback()
|
|
|
|
# Ofsted — latest inspection per URN. Ordered so the first row seen per
|
|
# URN is the most recent.
|
|
def _ofsted():
|
|
rows = (
|
|
db.query(FactOfstedInspection)
|
|
.filter(FactOfstedInspection.urn.in_(urns))
|
|
.order_by(FactOfstedInspection.urn, FactOfstedInspection.inspection_date.desc())
|
|
.all()
|
|
)
|
|
seen = set()
|
|
for o in rows:
|
|
if o.urn in seen:
|
|
continue
|
|
seen.add(o.urn)
|
|
result[o.urn]["ofsted"] = _ofsted_block(o, o.urn)
|
|
_safe(_ofsted)
|
|
|
|
# Census — latest year per URN.
|
|
def _census():
|
|
rows = (
|
|
db.query(FactPupilCharacteristics)
|
|
.filter(FactPupilCharacteristics.urn.in_(urns))
|
|
.order_by(FactPupilCharacteristics.urn, FactPupilCharacteristics.year.desc())
|
|
.all()
|
|
)
|
|
seen = set()
|
|
for pc in rows:
|
|
if pc.urn in seen:
|
|
continue
|
|
seen.add(pc.urn)
|
|
result[pc.urn]["census"] = _census_dict(pc)
|
|
_safe(_census)
|
|
|
|
# Admissions — all years per URN, oldest first (multi-year trend view).
|
|
def _admissions():
|
|
rows = (
|
|
db.query(FactAdmissions)
|
|
.filter(FactAdmissions.urn.in_(urns))
|
|
.order_by(FactAdmissions.urn, FactAdmissions.year.asc())
|
|
.all()
|
|
)
|
|
history: dict = {urn: [] for urn in urns}
|
|
for a in rows:
|
|
history[a.urn].append(_admissions_row_dict(a))
|
|
for urn, rows_for_urn in history.items():
|
|
result[urn]["admissions_history"] = rows_for_urn
|
|
result[urn]["admissions"] = rows_for_urn[-1] if rows_for_urn else None
|
|
_safe(_admissions)
|
|
|
|
# Last distance offered — the latest year per URN, and only that.
|
|
#
|
|
# The mart holds every published year and the DAG keeps loading them; what
|
|
# changed is what leaves this process. Earlier years are being held back as
|
|
# a paid feature, and this API is public and unauthenticated — serving the
|
|
# history here would hand it to anyone who opened the network tab, whatever
|
|
# the page chose to render. Withholding it in the client would have been
|
|
# decoration, not a decision.
|
|
#
|
|
# Restoring it for entitled callers is a change to this function, not to
|
|
# the pipeline: fact_admission_distance is untouched and complete.
|
|
def _admission_distance():
|
|
rows = (
|
|
db.query(FactAdmissionDistance)
|
|
.filter(FactAdmissionDistance.urn.in_(urns))
|
|
.order_by(FactAdmissionDistance.urn, FactAdmissionDistance.year.desc())
|
|
.all()
|
|
)
|
|
seen = set()
|
|
for d in rows:
|
|
if d.urn in seen:
|
|
continue
|
|
seen.add(d.urn)
|
|
result[d.urn]["admission_distance"] = _admission_distance_dict(d)
|
|
_safe(_admission_distance)
|
|
|
|
# Deprivation — one row per URN.
|
|
def _deprivation():
|
|
rows = (
|
|
db.query(FactDeprivation)
|
|
.filter(FactDeprivation.urn.in_(urns))
|
|
.all()
|
|
)
|
|
for d in rows:
|
|
result[d.urn]["deprivation"] = _deprivation_dict(d)
|
|
_safe(_deprivation)
|
|
|
|
# Finance — latest year per URN.
|
|
def _finance():
|
|
rows = (
|
|
db.query(FactFinance)
|
|
.filter(FactFinance.urn.in_(urns))
|
|
.order_by(FactFinance.urn, FactFinance.year.desc())
|
|
.all()
|
|
)
|
|
seen = set()
|
|
for f in rows:
|
|
if f.urn in seen:
|
|
continue
|
|
seen.add(f.urn)
|
|
result[f.urn]["finance"] = _finance_dict(f)
|
|
_safe(_finance)
|
|
|
|
# Destinations — KS4 and 16-18. Both marts are long-format, so every row
|
|
# for a URN is collected and _destinations_block picks the latest year and
|
|
# shapes the pupil groups. A phase with no rows serialises as null rather
|
|
# than an empty shell, so the frontend renders nothing rather than an empty
|
|
# section.
|
|
def _destinations():
|
|
from collections import defaultdict
|
|
|
|
def _collect(model):
|
|
per_urn = defaultdict(list)
|
|
for r in db.query(model).filter(model.urn.in_(urns)).all():
|
|
per_urn[r.urn].append({
|
|
"year": r.year,
|
|
"pupil_group": r.pupil_group,
|
|
"destination_measure": r.destination_measure,
|
|
"cohort_pupils": r.cohort_pupils,
|
|
"pupils": r.pupils,
|
|
"percentage": r.percentage,
|
|
"status": r.status,
|
|
})
|
|
return per_urn
|
|
|
|
ks4_rows = _collect(FactKs4Destinations)
|
|
ks5_rows = _collect(FactKs5Destinations)
|
|
for urn in urns:
|
|
ks4 = _destinations_block(ks4_rows.get(urn, []))
|
|
ks5 = _destinations_block(ks5_rows.get(urn, []))
|
|
result[urn]["destinations"] = (
|
|
{"ks4": ks4, "ks5": ks5} if (ks4 or ks5) else None
|
|
)
|
|
_safe(_destinations)
|
|
|
|
return result
|
|
|
|
|
|
def get_supplementary_data(db: Session, urn: int) -> dict:
|
|
"""Supplementary data for a single URN (thin wrapper over the batch)."""
|
|
return get_supplementary_data_batch(db, [urn])[int(urn)]
|