Merge pull request 'fix(pipeline): keep the destinations marts out of the scheduled builds' (#181) from fix/scheduled-dbt-selectors into main

Reviewed-on: #181

nextjs-app/__tests__/components/NearbySchoolsSection.test.tsx

@@ -8,7 +8,6 @@
 import { render, screen } from '@testing-library/react';
 
 import {
-  nearbyNoun,
   NearbySchoolsSection,
   shouldRenderNearby,
 } from '@/components/school/NearbySchoolsSection';
@@ -43,13 +42,7 @@ function school(overrides: Partial<NearbySchool> = {}): NearbySchool {
 
 function renderSection(nearby: NearbySchool[]) {
   return render(
-    <NearbySchoolsSection
-      urn={100001}
-      schoolName="Meadowbrook Primary School"
-      phase="Primary"
-      thisMetricValue={72}
-      nearby={nearby}
-    />,
+    <NearbySchoolsSection urn={100001} thisMetricValue={72} nearby={nearby} />,
   );
 }
 
@@ -90,15 +83,10 @@ describe('what the section claims', () => {
   });
 
   it('shows no chips at all when nothing is shared, rather than inventing one', () => {
-    const { container } = render(
-      <NearbySchoolsSection
-        urn={100001}
-        schoolName="Meadowbrook Primary School"
-        phase="Primary"
-        thisMetricValue={72}
-        nearby={[school({ shared: [] }), school({ urn: 100003, shared: [] })]}
-      />,
-    );
+    const { container } = renderSection([
+      school({ shared: [] }),
+      school({ urn: 100003, shared: [] }),
+    ]);
     // The card still carries its distance, name, type and figure — just no
     // claim of likeness.
     expect(container.querySelectorAll('li ul').length).toBe(0);
@@ -106,37 +94,6 @@ describe('what the section claims', () => {
   });
 });
 
-describe('what the lede calls the set', () => {
-  it.each([
-    ['Primary', 'primary schools'],
-    ['Middle deemed primary', 'primary schools'],
-    ['Secondary', 'secondary schools'],
-    ['Middle deemed secondary', 'secondary schools'],
-    ['All-through', 'all-through schools'],
-    // GIAS phase 6. Its candidates span the whole secondary group, so no
-    // single noun fits and it takes the honest general one.
-    ['16 plus', 'schools and colleges'],
-    ['', 'schools'],
-    [null, 'schools'],
-  ])('calls a %s school\'s neighbours "%s"', (phase, expected) => {
-    expect(nearbyNoun(phase)).toBe(expected);
-  });
-
-  it('never calls a sixth form college\'s neighbours primary schools', () => {
-    render(
-      <NearbySchoolsSection
-        urn={100001}
-        schoolName="Barnet Sixth Form College"
-        phase="16 plus"
-        thisMetricValue={null}
-        nearby={[school(), school({ urn: 100003 })]}
-      />,
-    );
-    expect(screen.getByText(/Other schools and colleges near Barnet Sixth Form College/)).toBeInTheDocument();
-    expect(screen.queryByText(/primary schools/)).not.toBeInTheDocument();
-  });
-});
-
 describe('cards', () => {
   it('links each school to its canonical slug', () => {
     renderSection([school(), school({ urn: 100003, school_name: 'Oakfield Primary School' })]);

nextjs-app/components/school/NearbySchools.module.css

@@ -1,8 +1,7 @@
 .heading { font-family: var(--font-display); font-size: 1.4rem; letter-spacing: -0.4px; margin: 0; }
-.lede { margin: 0.5rem 0 1.25rem; color: var(--text-secondary); max-width: 64ch; }
 .caption { margin: 1rem 0 0; font-size: 0.72rem; color: var(--text-muted); }
 
-.top { display: flex; align-items: flex-start; justify-content: space-between; gap: 1rem; }
+.top { display: flex; align-items: center; justify-content: space-between; gap: 1rem; margin-bottom: 1.25rem; }
 .arrows { display: flex; gap: 0.5rem; flex: none; }
 .arrow { width: 44px; height: 44px; display: grid; place-items: center; cursor: pointer; border: 1px solid var(--border-strong); border-radius: 999px; background: var(--bg-card); color: var(--brand); }
 .arrow:hover:not(:disabled) { border-color: var(--brand); background: var(--brand-bg); }
@@ -15,10 +14,10 @@
 .scroller { display: grid; grid-auto-flow: column; grid-auto-columns: calc((100% - 1.8rem) / 3); gap: 0.9rem; overflow-x: auto; scroll-snap-type: x mandatory; padding: 2px; margin: -2px; list-style: none; scrollbar-width: none; -ms-overflow-style: none; }
 .scroller::-webkit-scrollbar { display: none; }
 @media (max-width: 820px) { .scroller { grid-auto-columns: calc((100% - 0.9rem) / 2); } }
-/* Touch widths (MOBILE.md): the arrows would take 96px from a 328px card and
-   crush the lede into four lines, for a control swiping already provides. They
-   go, and the documented right-edge fade carries the affordance — lifting at
-   the end of the travel, where there is nothing more to hint at. */
+/* Touch widths (MOBILE.md): the arrows would take 96px from a 328px card, for
+   a control swiping already provides. They go, and the documented right-edge
+   fade carries the affordance — lifting at the end of the travel, where there
+   is nothing more to hint at. */
 @media (max-width: 640px) {
   .top { display: block; }
   .arrows { display: none; }

nextjs-app/components/school/NearbySchoolsCarousel.tsx

@@ -4,7 +4,7 @@
  * The scroller and its arrows.
  *
  * `children` are the server-rendered cards and `header` the server-rendered
- * heading and lede: both stay server components, passed through, so this file
+ * heading: both stay server components, passed through, so this file
  * owns a DOM ref and nothing else. That is what keeps all six links in the
  * initial HTML — a carousel that mounted cards on click would put four of the
  * six beyond a crawler and beyond a reader with no JavaScript.

nextjs-app/components/school/NearbySchoolsSection.tsx

@@ -13,8 +13,10 @@
  * reached on their behalf.
  *
  * There is deliberately no "how these are chosen" panel: the method is already
- * visible in the lede, the chips and the distances. The single caption line is
- * not a method note — it is the one thing a card cannot self-correct.
+ * visible in the chips and the distances. The single caption line is not a
+ * method note — it is the one thing a card cannot self-correct.
+ *
+ * Nor is there a lede: "Other primary schools near X" only restated the heading.
  */
 
 import Link from 'next/link';
@@ -32,28 +34,6 @@ export function shouldRenderNearby(nearby?: NearbySchool[] | null): boolean {
   return (nearby?.length ?? 0) >= MINIMUM;
 }
 
-/**
- * What the lede calls the set of schools it is showing.
- *
- * Derived from the school's own GIAS phase rather than the template it renders
- * with, because those disagree for "16 plus" (GIAS phase 6): a sixth-form
- * college renders the primary template — computeSchoolFlags tests for the
- * substring "secondary" — while the backend correctly matches it against the
- * secondary group. Taking the noun from the template would print "Other primary
- * schools near <sixth form college>" above a row of secondaries.
- *
- * A 16-plus school's candidates span the whole secondary group, so no single
- * noun fits and it gets the honest general one.
- */
-export function nearbyNoun(phase: string | null | undefined): string {
-  const text = (phase ?? '').trim().toLowerCase();
-  if (text === 'all-through') return 'all-through schools';
-  if (text === '16 plus') return 'schools and colleges';
-  if (text.includes('secondary')) return 'secondary schools';
-  if (text.includes('primary')) return 'primary schools';
-  return 'schools';
-}
-
 function metricLabel(key: string): string {
   return key === 'attainment_8_score' ? 'Attainment 8' : 'Reading, writing & maths';
 }
@@ -65,25 +45,16 @@ function formatMetric(value: number | null, key: string): string {
 
 export function NearbySchoolsSection({
   urn,
-  schoolName,
-  phase,
   thisMetricValue,
   nearby,
 }: {
   urn: number;
-  schoolName: string;
-  /** The school's own GIAS phase, not the template it renders with. */
-  phase: string | null | undefined;
   thisMetricValue: number | null;
   nearby?: NearbySchool[] | null;
 }) {
   if (!shouldRenderNearby(nearby)) return null;
   const schools = nearby as NearbySchool[];
-
-  // One card matched on phase alone, so the section may not claim the set
-  // shares an intake with this school.
   const metricKey = schools[0].metric_key;
-  const noun = nearbyNoun(phase);
 
   return (
     <Section id="nearby">
@@ -91,12 +62,9 @@ export function NearbySchoolsSection({
         count={schools.length}
         labelledBy="nearby-schools-heading"
         header={
-          <div>
-            <h2 id="nearby-schools-heading" className={styles.heading}>
-              Other schools nearby
-            </h2>
-            <p className={styles.lede}>{`Other ${noun} near ${schoolName}.`}</p>
-          </div>
+          <h2 id="nearby-schools-heading" className={styles.heading}>
+            Other schools nearby
+          </h2>
         }
       >
         {schools.map((school) => (

nextjs-app/components/school/PrimarySchoolSections.tsx

@@ -154,8 +154,6 @@ export function PrimarySchoolSections({
       {/* Last: it is where the reader goes next, not part of this school. */}
       <NearbySchoolsSection
         urn={schoolInfo.urn}
-        schoolName={schoolInfo.school_name}
-        phase={schoolInfo.phase}
         thisMetricValue={flags.latestResults?.rwm_expected_pct ?? null}
         nearby={nearbySchools}
       />

nextjs-app/components/school/SecondarySchoolSections.tsx

@@ -148,8 +148,6 @@ export function SecondarySchoolSections({
       {/* Last: it is where the reader goes next, not part of this school. */}
       <NearbySchoolsSection
         urn={schoolInfo.urn}
-        schoolName={schoolInfo.school_name}
-        phase={schoolInfo.phase}
         thisMetricValue={flags.latestResults?.attainment_8_score ?? null}
         nearby={nearbySchools}
       />

pipeline/dags/school_data_pipeline.py

@@ -106,9 +106,12 @@ print(f'Validation passed: {{count}} GIAS rows')
         """,
     )
 
+    # Marts fed by annual EES staging models are rebuilt by the EES DAG, even
+    # when they join dim_school. Selecting them here fails in any database
+    # where that DAG hasn't run (pipeline/tests/test_dag_selectors.py).
     dbt_build = BashOperator(
         task_id="dbt_build",
-        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_gias_establishments+ stg_gias_links+ gias_code_names+ --exclude int_ks2_with_lineage+ int_ks4_with_lineage+",
+        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_gias_establishments+ stg_gias_links+ gias_code_names+ --exclude int_ks2_with_lineage+ int_ks4_with_lineage+ stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
     )
 
     sync_typesense = BashOperator(
@@ -143,7 +146,7 @@ with DAG(
 
     dbt_build_ofsted = BashOperator(
         task_id="dbt_build",
-        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_ofsted_inspections+ int_ofsted_latest+ fact_ofsted_inspection+ dim_school+",
+        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_ofsted_inspections+ int_ofsted_latest+ fact_ofsted_inspection+ dim_school+ --exclude stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
     )
 
     sync_typesense_ofsted = BashOperator(

pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/gias_csv.py

@@ -1,26 +0,0 @@
-"""Read a GIAS extract from the raw bytes of the download.
-
-GIAS writes its CSVs in Windows-1252 and sends no charset, so `resp.text`
-leaves requests to guess the codec. On 3 Oct 2026 it guessed windows-1250 and
-"à" became "ŕ". Decode the bytes ourselves instead.
-"""
-
-from __future__ import annotations
-
-import io
-
-import pandas as pd
-
-GIAS_ENCODING = "cp1252"
-
-
-def read_gias_csv(content: bytes, logger=None) -> pd.DataFrame:
-    """Every column as a string; a blank cell stays ''."""
-    # Windows-1252 leaves five bytes undefined. One stray byte must not stop
-    # the daily refresh of every school, so it becomes U+FFFD and is logged.
-    text = content.decode(GIAS_ENCODING, errors="replace")
-    undecodable = text.count("�")
-    if undecodable and logger is not None:
-        logger.warning("%d byte(s) in the GIAS extract could not be decoded as %s",
-                       undecodable, GIAS_ENCODING)
-    return pd.read_csv(io.StringIO(text), dtype=str, keep_default_na=False)

pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py

@@ -7,8 +7,6 @@ from datetime import date, timedelta
 from singer_sdk import Stream, Tap
 from singer_sdk import typing as th
 
-from tap_uk_gias.gias_csv import read_gias_csv
-
 GIAS_URL_TEMPLATE = (
     "https://ea-edubase-api-prod.azurewebsites.net"
     "/edubase/downloads/public/edubasealldata{date}.csv"
@@ -76,6 +74,9 @@ class GIASEstablishmentsStream(Stream):
 
     def get_records(self, context):
         """Download GIAS CSV and yield rows."""
+        import io
+
+        import pandas as pd
         import requests
 
         today = date.today()
@@ -93,7 +94,12 @@ class GIASEstablishmentsStream(Stream):
 
         resp.raise_for_status()
 
-        df = read_gias_csv(resp.content, self.logger)
+        df = pd.read_csv(
+            io.StringIO(resp.text),
+            encoding="latin-1",
+            dtype=str,
+            keep_default_na=False,
+        )
 
         for _, row in df.iterrows():
             record = row.to_dict()
@@ -120,6 +126,9 @@ class GIASLinksStream(Stream):
 
     def get_records(self, context):
         """Download GIAS links CSV and yield rows."""
+        import io
+
+        import pandas as pd
         import requests
 
         today = date.today()
@@ -137,7 +146,12 @@ class GIASLinksStream(Stream):
 
         resp.raise_for_status()
 
-        df = read_gias_csv(resp.content, self.logger)
+        df = pd.read_csv(
+            io.StringIO(resp.text),
+            encoding="latin-1",
+            dtype=str,
+            keep_default_na=False,
+        )
 
         for _, row in df.iterrows():
             record = row.to_dict()

pipeline/tests/test_dag_selectors.py

@@ -0,0 +1,98 @@
+"""Every scheduled dbt build must only build models whose parents exist.
+
+The daily GIAS build selects `stg_gias_establishments+`, so any mart that joins
+dim_school joins the daily build too. When such a mart also reads a staging
+model that only the manually triggered EES DAG builds, the daily build fails in
+any database where that DAG has not run since. Sync and cache invalidation then
+never run either. The destinations marts did this from late August 2026.
+
+The graph is read from the model SQL, because CI has no dbt.
+"""
+import re
+from collections import defaultdict
+from pathlib import Path
+
+import pytest
+
+PIPELINE = Path(__file__).resolve().parents[1]
+MODELS = PIPELINE / 'transform' / 'models'
+DAG_FILE = PIPELINE / 'dags' / 'school_data_pipeline.py'
+
+REF = re.compile(r"ref\(\s*'([a-z0-9_]+)'\s*\)")
+DBT_BUILD = re.compile(r'dbt_build\w*\s*=\s*BashOperator\(.*?build --profiles-dir \. --target production ([^"]+)"', re.S)
+DAG_ID = re.compile(r'dag_id="([a-z0-9_]+)"')
+
+DAILY = 'school_data_daily'
+
+# dim_school reads int_ofsted_latest only when the relation exists
+# (adapter.get_relation), so a missing table is not a failure.
+OPTIONAL_PARENTS = {'int_ofsted_latest'}
+
+
+def model_parents():
+    """{model: models it refs}. Seeds are left out: they are loaded once and always exist."""
+    sql = {p.stem: p.read_text() for p in MODELS.rglob('*.sql')}
+    return {name: set(REF.findall(text)) & set(sql) for name, text in sql.items()}
+
+
+def downstream(node, children):
+    seen, stack = {node}, [node]
+    while stack:
+        for child in children[stack.pop()]:
+            if child not in seen:
+                seen.add(child)
+                stack.append(child)
+    return seen
+
+
+def expand(tokens, children):
+    out = set()
+    for token in tokens:
+        out |= downstream(token[:-1], children) if token.endswith('+') else {token}
+    return out
+
+
+def scheduled_builds():
+    """{dag_id: dbt selection arguments} for every dbt build in the DAG file."""
+    text = DAG_FILE.read_text()
+    starts = [(m.start(), m.group(1)) for m in DAG_ID.finditer(text)]
+    builds = {}
+    for i, (start, dag_id) in enumerate(starts):
+        end = starts[i + 1][0] if i + 1 < len(starts) else len(text)
+        found = DBT_BUILD.search(text, start, end)
+        if found:
+            builds[dag_id] = found.group(1)
+    return builds
+
+
+def selected_models(args, parents):
+    children = defaultdict(set)
+    for model, ps in parents.items():
+        for p in ps:
+            children[p].add(model)
+    select = re.search(r'--select (.+?)(?= --exclude|$)', args).group(1).split()
+    excluded = re.search(r'--exclude (.+)$', args)
+    exclude = excluded.group(1).split() if excluded else []
+    return (expand(select, children) - expand(exclude, children)) & set(parents)
+
+
+PARENTS = model_parents()
+BUILDS = scheduled_builds()
+DAILY_MODELS = selected_models(BUILDS[DAILY], PARENTS)
+
+
+def test_every_dag_with_a_dbt_build_is_parsed():
+    assert set(BUILDS) == {
+        'school_data_daily', 'school_data_monthly_ofsted', 'school_data_annual_ees',
+        'school_data_annual_idaci', 'school_data_annual_distance',
+    }
+
+
+@pytest.mark.parametrize('dag_id', sorted(BUILDS))
+def test_selected_models_only_read_models_that_exist(dag_id):
+    selected = selected_models(BUILDS[dag_id], PARENTS)
+    # The daily build is the base layer: other DAGs may rely on what it builds.
+    available = selected | OPTIONAL_PARENTS | (DAILY_MODELS if dag_id != DAILY else set())
+    missing = {model: sorted(PARENTS[model] - available) for model in sorted(selected)
+               if PARENTS[model] - available}
+    assert missing == {}, f'{dag_id} builds models whose parents it never builds: {missing}'

pipeline/tests/test_gias_encoding.py

@@ -1,66 +0,0 @@
-"""GIAS publishes its extracts in Windows-1252 and declares no charset.
-
-The tap used to hand pandas `resp.text`, so requests guessed the codec.
-On 3 Oct 2026 it guessed windows-1250, and "St Thomas à Becket" was stored
-as "St Thomas ŕ Becket". The `encoding=` passed to read_csv did nothing,
-because the text was already decoded.
-"""
-import importlib.util
-import logging
-from pathlib import Path
-
-import pytest
-
-MODULE = (Path(__file__).resolve().parents[1] / 'plugins' / 'extractors' / 'tap-uk-gias'
-          / 'tap_uk_gias' / 'gias_csv.py')
-
-
-@pytest.fixture
-def gias_csv():
-    spec = importlib.util.spec_from_file_location('gias_csv', MODULE)
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
-# Byte for byte as GIAS writes it: 0xE0 à, 0x92 ’, 0xE9 é, 0xB0 °, 0xE7 ç.
-EXTRACT = (
-    b'"URN","EstablishmentName","HeadLastName"\r\n'
-    b'"138950","St Thomas \xe0 Becket Catholic Secondary School","Smith"\r\n'
-    b'"100000","The Dean and Chapter of St Paul\x92s Cathedral","Pr\xe9vert"\r\n'
-    b'"140677","North Star 180\xb0","Fran\xe7ois"\r\n'
-    b'"100001","No head recorded",""\r\n'
-)
-
-
-def test_names_decode_as_windows_1252(gias_csv):
-    df = gias_csv.read_gias_csv(EXTRACT)
-    assert list(df['EstablishmentName']) == [
-        'St Thomas à Becket Catholic Secondary School',
-        'The Dean and Chapter of St Paul’s Cathedral',
-        'North Star 180°',
-        'No head recorded',
-    ]
-    assert list(df['HeadLastName']) == ['Smith', 'Prévert', 'François', '']
-
-
-def test_the_codec_requests_guessed_is_not_used(gias_csv):
-    # What the tap stored on 3 Oct: the same bytes read as windows-1250.
-    assert 'ŕ' in EXTRACT.decode('cp1250')
-    names = ' '.join(gias_csv.read_gias_csv(EXTRACT)['EstablishmentName'])
-    assert 'ŕ' not in names
-
-
-def test_values_stay_strings(gias_csv):
-    df = gias_csv.read_gias_csv(EXTRACT)
-    assert df.loc[0, 'URN'] == '138950'
-
-
-def test_a_byte_windows_1252_leaves_undefined_does_not_stop_the_load(gias_csv, caplog):
-    # 0x81 has no Windows-1252 character. One odd name must not block the daily
-    # refresh of every school, but it must be visible in the log.
-    extract = b'"URN","EstablishmentName"\r\n"100002","Odd \x81 Name"\r\n'
-    with caplog.at_level(logging.WARNING):
-        df = gias_csv.read_gias_csv(extract, logger=logging.getLogger('gias'))
-    assert df.loc[0, 'EstablishmentName'] == 'Odd � Name'
-    assert 'could not be decoded' in caplog.text