feat(destinations): serve destinations without closing the gaps

The serialiser carries status through and computes no totals of its own.
The only aggregates in the payload are ones DfE published itself; whether
showing one is safe depends on how many of its components are suppressed,
which the frontend decides.

The batch guard grows from six tables to eight.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BvdDKvFFSZuMVDH5fEyTob

backend/data_loader.py

@@ -20,6 +20,7 @@ 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
@@ -816,6 +817,73 @@ def _finance_dict(f) -> dict:
     }
 
 
+# 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
+
+
+def _destinations_block(rows: list) -> dict | None:
+    """Shape destination rows for one phase into the API's block.
+
+    Carries `status` through untouched and emits no computed totals. The only
+    aggregates present are ones DfE published itself; whether showing one is
+    safe depends on how many of its components are suppressed, which the
+    frontend decides (lib/destinations.ts, rule R2).
+
+    Deliberately does NOT compute a residual, a "remaining pupils" figure, or
+    any total that would close a gap left by a suppressed category — the
+    categories sum to the cohort, so such a figure names the withheld cell.
+    """
+    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": [], "aggregates": {}},
+        )
+        measure = row["destination_measure"]
+        cell = {
+            "category": measure,
+            "pupils": row.get("pupils"),
+            "percentage": row.get("percentage"),
+            "status": row.get("status"),
+        }
+        if measure in _AGGREGATE_MEASURES:
+            group["aggregates"][measure[len("agg_"):]] = cell
+        else:
+            group["categories"].append(cell)
+
+    if not groups:
+        return None
+
+    return {"cohort_year": _format_cohort_year(latest_year), "groups": groups}
+
+
 def _empty_supplementary() -> dict:
     return {
         "ofsted": None,
@@ -827,6 +895,7 @@ def _empty_supplementary() -> dict:
         "phonics": None,
         "deprivation": None,
         "finance": None,
+        "destinations": None,
     }
 
 
@@ -954,6 +1023,38 @@ def get_supplementary_data_batch(db: Session, urns: list[int]) -> dict:
             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