Files
school_compare/docs/superpowers/plans/2026-07-09-gias-code-dictionaries.md
2026-07-09 10:10:26 +01:00

34 KiB

GIAS Code Dictionaries Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Store the six GIAS classification fields as official DfE integer codes in the marts and translate code → name in application code, leaving the API contract (name strings) unchanged.

Architecture: A generation script downloads the public GIAS bulk CSV and emits the dictionaries (Python dicts + a dbt seed) from real data. The tap ingests the (code) columns, staging casts them, dim_school/dim_location keep only codes, and translation happens in exactly two places: backend/data_loader.py right after pd.read_sql, and pipeline/scripts/sync_typesense.py before indexing. A dbt seed test warns when DfE adds/renames a value; a parity test keeps the backend and pipeline dictionary copies identical.

Tech Stack: Singer SDK tap, dbt (Postgres), FastAPI + pandas, Typesense sync script, pytest.

Spec: docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md

Global Constraints

  • Numeric code values are never assumed. Every literal code used in SQL or yml (status filter, sixth-form derivation, phase cascade) must be verified against pipeline/transform/seeds/gias_code_names.csv generated in Task 1 from the live CSV. The literals written in this plan are best-current-knowledge and each carries a verification step.
  • Names served by the API must stay byte-identical to today's strings (e.g. Does not apply, Open, but proposed to close) — UI heuristics compare exact strings.
  • The (name) columns stay declared in the tap and present in raw; staging stops exposing them.
  • dim_school and dim_location status filters must stay identical (API inner-joins them).
  • Backend tests run via: uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests -v (no local pytest exists).
  • dbt cannot run locally — dbt changes are verified statically (grep / yaml parse) + CI.
  • Never push to main. Work on branch feat/gias-code-dictionaries (branch off docs/gias-code-dictionaries so the spec is included, or off main if that has merged).
  • Commits end with: Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
  • Deploy runbook (accepted window, spec §7): merge → deploy → trigger school_data_daily immediately. No code-level fallback for the old-schema window.

Task 1: Dictionary generation script, canonical module, pipeline copy, seed

Files:

  • Create: pipeline/scripts/generate_gias_codes.py
  • Create: backend/gias_codes.py (content generated by the script)
  • Create: pipeline/scripts/gias_codes.py (byte-identical copy)
  • Create: pipeline/transform/seeds/gias_code_names.csv (generated)
  • Test: backend/tests/test_gias_codes.py

Interfaces:

  • Produces: backend/gias_codes.py exporting SCHOOL_TYPE, ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION, OFFICIAL_SIXTH_FORM, RELIGIOUS_CHARACTER, ADMISSIONS_POLICY (each dict[int, str]) and translate(code, mapping) -> str | None. Task 4 imports these; Task 5 imports the pipeline copy; Task 3 reads code literals from the seed CSV.

  • Step 1: Write the failing tests

Create backend/tests/test_gias_codes.py:

"""Tests for the GIAS code->name dictionaries (spec 2026-07-09).

The dictionaries are generated from the live GIAS bulk CSV by
pipeline/scripts/generate_gias_codes.py — these tests assert the module's
contract, key sentinel values the marts/UI depend on, and that the pipeline
copy has not drifted from the canonical backend module.
"""

import math
from pathlib import Path

from backend.gias_codes import (
    ADMISSIONS_POLICY,
    ESTABLISHMENT_STATUS,
    OFFICIAL_SIXTH_FORM,
    PHASE_OF_EDUCATION,
    RELIGIOUS_CHARACTER,
    SCHOOL_TYPE,
    translate,
)

REPO = Path(__file__).resolve().parents[2]


def test_translate_known_code():
    open_code = next(c for c, n in ESTABLISHMENT_STATUS.items() if n == "Open")
    assert translate(open_code, ESTABLISHMENT_STATUS) == "Open"


def test_translate_unknown_code_degrades_gracefully():
    assert translate(9999, ESTABLISHMENT_STATUS) == "Unknown (9999)"


def test_translate_none_and_nan_return_none():
    assert translate(None, ESTABLISHMENT_STATUS) is None
    assert translate(float("nan"), ESTABLISHMENT_STATUS) is None


def test_translate_accepts_float_codes():
    # pd.read_sql yields float columns when NULLs are present
    open_code = next(c for c, n in ESTABLISHMENT_STATUS.items() if n == "Open")
    assert translate(float(open_code), ESTABLISHMENT_STATUS) == "Open"


def test_sentinel_names_present():
    """Names the marts/UI compare against must exist verbatim."""
    assert "Open" in ESTABLISHMENT_STATUS.values()
    assert "Open, but proposed to close" in ESTABLISHMENT_STATUS.values()
    assert "Has a sixth form" in OFFICIAL_SIXTH_FORM.values()
    assert "Primary" in PHASE_OF_EDUCATION.values()
    assert "Secondary" in PHASE_OF_EDUCATION.values()
    assert "Does not apply" in RELIGIOUS_CHARACTER.values()
    assert all(len(d) > 0 for d in (
        SCHOOL_TYPE, ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION,
        OFFICIAL_SIXTH_FORM, RELIGIOUS_CHARACTER, ADMISSIONS_POLICY,
    ))


def test_pipeline_copy_is_identical():
    canonical = (REPO / "backend" / "gias_codes.py").read_text()
    copy = (REPO / "pipeline" / "scripts" / "gias_codes.py").read_text()
    assert canonical == copy, (
        "pipeline/scripts/gias_codes.py has drifted from backend/gias_codes.py — "
        "regenerate with pipeline/scripts/generate_gias_codes.py and copy the file"
    )


def test_seed_matches_dictionaries():
    import csv
    fields = {
        "school_type": SCHOOL_TYPE,
        "establishment_status": ESTABLISHMENT_STATUS,
        "phase_of_education": PHASE_OF_EDUCATION,
        "official_sixth_form": OFFICIAL_SIXTH_FORM,
        "religious_character": RELIGIOUS_CHARACTER,
        "admissions_policy": ADMISSIONS_POLICY,
    }
    seed_path = REPO / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
    seed: dict[str, dict[int, str]] = {k: {} for k in fields}
    with open(seed_path, newline="") as fh:
        for row in csv.DictReader(fh):
            seed[row["field"]][int(row["code"])] = row["name"]
    assert seed == fields
  • Step 2: Run tests to verify they fail

Run: cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests/test_gias_codes.py -v Expected: FAIL at import — ModuleNotFoundError: No module named 'backend.gias_codes'.

  • Step 3: Write the generation script

Create pipeline/scripts/generate_gias_codes.py:

"""Generate GIAS code->name dictionaries from the live bulk CSV.

Writes:
  - backend/gias_codes.py                        (canonical Python module)
  - pipeline/scripts/gias_codes.py               (byte-identical copy)
  - pipeline/transform/seeds/gias_code_names.csv (dbt seed for drift test)

Run from the repo root whenever the dbt drift test warns that DfE
added/renamed a value:  python pipeline/scripts/generate_gias_codes.py
"""

from __future__ import annotations

import io
import sys
from datetime import date, timedelta
from pathlib import Path

import pandas as pd
import requests

GIAS_URL = (
    "https://ea-edubase-api-prod.azurewebsites.net"
    "/edubase/downloads/public/edubasealldata{date}.csv"
)

# (CSV code column, CSV name column, python dict name, seed field key)
FIELDS = [
    ("TypeOfEstablishment (code)", "TypeOfEstablishment (name)", "SCHOOL_TYPE", "school_type"),
    ("EstablishmentStatus (code)", "EstablishmentStatus (name)", "ESTABLISHMENT_STATUS", "establishment_status"),
    ("PhaseOfEducation (code)", "PhaseOfEducation (name)", "PHASE_OF_EDUCATION", "phase_of_education"),
    ("OfficialSixthForm (code)", "OfficialSixthForm (name)", "OFFICIAL_SIXTH_FORM", "official_sixth_form"),
    ("ReligiousCharacter (code)", "ReligiousCharacter (name)", "RELIGIOUS_CHARACTER", "religious_character"),
    ("AdmissionsPolicy (code)", "AdmissionsPolicy (name)", "ADMISSIONS_POLICY", "admissions_policy"),
]

MODULE_HEADER = '''"""GIAS code -> name dictionaries.

GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
— do not edit by hand; rerun the script when the dbt drift test warns.
The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
must be byte-identical (enforced by backend/tests/test_gias_codes.py).
"""

from __future__ import annotations

import logging
import math

logger = logging.getLogger(__name__)

'''

MODULE_FOOTER = '''

def translate(code, mapping: dict[int, str]) -> str | None:
    """Translate a GIAS code to its display name.

    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
    """
    if code is None or (isinstance(code, float) and math.isnan(code)):
        return None
    code = int(code)
    if code not in mapping:
        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
        return f"Unknown ({code})"
    return mapping[code]
'''


def download_csv() -> pd.DataFrame:
    for day in (date.today(), date.today() - timedelta(days=1)):
        url = GIAS_URL.format(date=day.strftime("%Y%m%d"))
        print(f"Downloading {url}")
        resp = requests.get(url, timeout=300)
        if resp.status_code == 404:
            continue
        resp.raise_for_status()
        return pd.read_csv(
            io.StringIO(resp.content.decode("latin-1")),
            dtype=str, keep_default_na=False,
        )
    sys.exit("GIAS CSV not available for today or yesterday")


def main() -> None:
    repo = Path(__file__).resolve().parents[2]
    df = download_csv()

    module_parts = [MODULE_HEADER]
    seed_rows: list[tuple[str, int, str]] = []

    for code_col, name_col, dict_name, field_key in FIELDS:
        pairs = (
            df[[code_col, name_col]]
            .loc[lambda d: (d[code_col] != "") & (d[name_col] != "")]
            .drop_duplicates()
        )
        mapping = sorted((int(c), n) for c, n in pairs.itertuples(index=False))
        dupes = len(mapping) - len({c for c, _ in mapping})
        if dupes:
            sys.exit(f"{code_col}: {dupes} codes map to multiple names — investigate before generating")
        lines = [f"{dict_name}: dict[int, str] = {{"]
        for code, name in mapping:
            escaped = name.replace('"', '\\"')
            lines.append(f'    {code}: "{escaped}",')
        lines.append("}\n")
        module_parts.append("\n".join(lines))
        seed_rows += [(field_key, code, name) for code, name in mapping]

    module = "\n".join(module_parts) + MODULE_FOOTER

    (repo / "backend" / "gias_codes.py").write_text(module)
    (repo / "pipeline" / "scripts" / "gias_codes.py").write_text(module)

    seed_path = repo / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
    with open(seed_path, "w", newline="") as fh:
        import csv
        w = csv.writer(fh)
        w.writerow(["field", "code", "name"])
        w.writerows(seed_rows)

    print(f"Wrote backend/gias_codes.py, pipeline/scripts/gias_codes.py, {seed_path.name}")
    print("\nKey codes for the dbt work (Task 3):")
    for field in ("establishment_status", "phase_of_education", "official_sixth_form"):
        print(f"  {field}:")
        for f, code, name in seed_rows:
            if f == field:
                print(f"    {code} = {name}")


if __name__ == "__main__":
    main()
  • Step 4: Run the generator

Run: cd /Users/tudor/projects/school_compare && uv run --with pandas --with requests python pipeline/scripts/generate_gias_codes.py Expected: downloads the CSV (~100MB, may take a minute), writes the three files, and prints the status/phase/sixth-form code tables. Record the printed code tables — Task 3 needs them. If the download fails twice, report BLOCKED (no network or GIAS outage) rather than inventing dictionary content.

  • Step 5: Run the tests again

Run: cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests/test_gias_codes.py -v Expected: 7 passed. If test_sentinel_names_present fails, the GIAS vocabulary differs from expectations — inspect the generated module and report DONE_WITH_CONCERNS naming the differing value; do not edit the generated names.

  • Step 6: Commit
git add pipeline/scripts/generate_gias_codes.py backend/gias_codes.py pipeline/scripts/gias_codes.py pipeline/transform/seeds/gias_code_names.csv backend/tests/test_gias_codes.py
git commit -m "feat: GIAS code->name dictionaries generated from live bulk CSV

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

Task 2: Tap ingests the (code) columns; staging exposes codes, drops names

Files:

  • Modify: pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py (Singer schema)
  • Modify: pipeline/transform/models/staging/stg_gias_establishments.sql

Interfaces:

  • Produces: staging columns school_type_code, status_code, phase_code, official_sixth_form_code, religious_character_code, admissions_policy_code (all int) consumed by Task 3. Staging stops exposing school_type, status, phase, official_sixth_form, religious_character, admissions_policy (names stay in raw only).

  • Step 1: Add the six (code) properties to the Singer schema

In tap.py, GIASEstablishmentsStream.schema, add each (code) property directly above its existing (name) sibling:

        th.Property("TypeOfEstablishment (code)", th.StringType),
        th.Property("PhaseOfEducation (code)", th.StringType),
        th.Property("EstablishmentStatus (code)", th.StringType),
        th.Property("Gender (name)", ...)  # existing line — for placement reference only
        th.Property("ReligiousCharacter (code)", th.StringType),
        th.Property("AdmissionsPolicy (code)", th.StringType),
        th.Property("OfficialSixthForm (code)", th.StringType),

(The exact insertion order doesn't matter — the schema is a dict — but keep each (code) adjacent to its (name) for readability. Do NOT remove any (name) property.)

  • Step 2: Rewrite the six columns in staging

In stg_gias_establishments.sql renamed CTE, replace:

        "TypeOfEstablishment (name)"                    as school_type,
        "PhaseOfEducation (name)"                       as phase,
        nullif(trim("OfficialSixthForm (name)"), '')    as official_sixth_form,
        "ReligiousCharacter (name)"                     as religious_character,
        "AdmissionsPolicy (name)"                       as admissions_policy,
        "EstablishmentStatus (name)"                    as status,

with:

        cast(nullif(trim("TypeOfEstablishment (code)"), '') as integer)  as school_type_code,
        cast(nullif(trim("PhaseOfEducation (code)"), '') as integer)     as phase_code,
        cast(nullif(trim("OfficialSixthForm (code)"), '') as integer)    as official_sixth_form_code,
        cast(nullif(trim("ReligiousCharacter (code)"), '') as integer)   as religious_character_code,
        cast(nullif(trim("AdmissionsPolicy (code)"), '') as integer)     as admissions_policy_code,
        cast(nullif(trim("EstablishmentStatus (code)"), '') as integer)  as status_code,

(The name lines are scattered through the CTE — replace each in place; the six name aliases must no longer appear in the model.)

  • Step 3: Verify statically

Run:

cd /Users/tudor/projects/school_compare && \
python3 -c "import ast; ast.parse(open('pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py').read()); print('tap OK')" && \
grep -c "(code)" pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py && \
grep -E "as (school_type|status|phase|official_sixth_form|religious_character|admissions_policy)," pipeline/transform/models/staging/stg_gias_establishments.sql; echo "name-alias grep exit=$? (want 1 = none found)"

Expected: tap OK, code-column count 6, and the final grep finds nothing (exit 1).

  • Step 4: Commit
git add pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py pipeline/transform/models/staging/stg_gias_establishments.sql
git commit -m "feat(pipeline): ingest GIAS code columns; staging exposes codes not names

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

Task 3: Marts store codes; dbt tests + drift test

Files:

  • Modify: pipeline/transform/models/marts/dim_school.sql
  • Modify: pipeline/transform/models/marts/dim_location.sql
  • Modify: pipeline/transform/models/marts/_marts_schema.yml
  • Create: pipeline/transform/tests/assert_gias_code_names_match_seed.sql

Interfaces:

  • Consumes: staging code columns from Task 2; code literals from pipeline/transform/seeds/gias_code_names.csv (Task 1).
  • Produces: dim_school columns school_type_code, status_code, phase_code, religious_character_code, admissions_policy_code (int) replacing their string columns; has_sixth_form unchanged (bool). Task 4's _MAIN_QUERY selects these.

Before writing SQL: open pipeline/transform/seeds/gias_code_names.csv and confirm the literals below. Best-current-knowledge values (VERIFY EACH): establishment_status: 1 = Open, 3 = "Open, but proposed to close" (2 = Closed, 4 = Proposed to open). phase_of_education: 0 = Not applicable, 2 = Primary, 4 = Secondary, 7 = All-through. official_sixth_form: 1 = Has a sixth form, 2 = Does not have a sixth form, 0 = Not applicable. If any differ, use the seed's values everywhere below and say so in your report.

  • Step 1: Rewrite dim_school.sql derivations in code space

Replace the phase cascade block (case ... end as phase,) with:

    -- Phase in GIAS code space (see seeds/gias_code_names.csv):
    -- 2 = Primary, 4 = Secondary, 7 = All-through, 0 = Not applicable.
    case
        -- 1. Trust GIAS phase when it's a real value (0 = the catch-all "Not Applicable")
        when s.phase_code is not null and s.phase_code != 0
            then s.phase_code
        -- 2. Infer from statutory age range (independent schools still publish these)
        when s.statutory_high_age is not null and s.statutory_high_age <= 11 then 2
        when s.statutory_low_age is not null and s.statutory_low_age >= 11 then 4
        when s.statutory_low_age is not null and s.statutory_high_age is not null
             and s.statutory_low_age < 11 and s.statutory_high_age > 11 then 7
        -- 3. Fallback: infer from school name (covers independents with missing ages)
        when s.school_name ilike '%primary%'
             or s.school_name ilike '%infant%'
             or s.school_name ilike '%junior%'
             or s.school_name ilike '%preparatory%'
             or s.school_name ilike '% prep school%'
             or s.school_name ilike '% prep %'
            then 2
        when s.school_name ilike '%secondary%'
             or s.school_name ilike '%high school%'
             or s.school_name ilike '%grammar%'
             or s.school_name ilike '%senior school%'
             or s.school_name ilike '%upper school%'
            then 4
        -- 4. Give up — null renders no phase pill
        else null
    end as phase_code,

Replace s.school_type, with s.school_type_code,; s.religious_character, with s.religious_character_code,; s.admissions_policy, with s.admissions_policy_code,; s.status, with s.status_code,.

Replace the has_sixth_form case with:

    -- GIAS OfficialSixthForm in code space: 1 = has, 2 = does not, 0 = N/A.
    -- Null (rare, new establishments) falls back to the statutory age range.
    case
        when s.official_sixth_form_code = 1 then true
        when s.official_sixth_form_code in (0, 2) then false
        else coalesce(s.statutory_high_age >= 18, false)
    end as has_sixth_form,

Replace the status filter with:

-- 1 = Open; 3 = Open, but proposed to close (still operating; drops out when
-- GIAS flips to Closed — marts fully rebuild each run).
where s.status_code in (1, 3)
  • Step 2: Same filter in dim_location.sql

Replace its where s.status in ('Open', 'Open, but proposed to close') (and the comment above it) with:

-- Must match dim_school's status filter exactly (the API inner-joins the two).
where s.status_code in (1, 3)
  • Step 3: Update _marts_schema.yml

Under dim_school columns: rename phasephase_code (keep the warn-severity not_null, reword description to mention codes); replace the status accepted_values block with:

      - name: status_code
        description: GIAS EstablishmentStatus code (1 = Open, 3 = Open but proposed to close)
        tests:
          - accepted_values:
              values: [1, 3]

Add warn-severity accepted_values for the other codes, values copied from the seed (school_type/religious/admissions lists are long — paste the full code list from gias_code_names.csv for each):

      - name: school_type_code
        tests:
          - accepted_values:
              severity: warn
              values: [<all school_type codes from the seed>]
      - name: religious_character_code
        tests:
          - accepted_values:
              severity: warn
              values: [<all religious_character codes from the seed>]
      - name: admissions_policy_code
        tests:
          - accepted_values:
              severity: warn
              values: [<all admissions_policy codes from the seed>]

(<...> here means: paste the actual comma-separated integers from the seed file — the lists exist by the time this task runs. Leaving a literal <...> in the yml is a task failure.)

has_sixth_form tests stay unchanged.

  • Step 4: Write the drift test

Create pipeline/transform/tests/assert_gias_code_names_match_seed.sql:

-- Warn when the live GIAS CSV carries a (code, name) pair we don't have in
-- the dictionary seed — i.e. DfE added or renamed a value. Fix by rerunning
-- pipeline/scripts/generate_gias_codes.py and committing the regenerated
-- dictionaries + seed together.
{{ config(severity='warn') }}

with raw_pairs as (
    {% for field_key, code_col, name_col in [
        ('school_type',          'TypeOfEstablishment (code)',  'TypeOfEstablishment (name)'),
        ('establishment_status', 'EstablishmentStatus (code)',  'EstablishmentStatus (name)'),
        ('phase_of_education',   'PhaseOfEducation (code)',     'PhaseOfEducation (name)'),
        ('official_sixth_form',  'OfficialSixthForm (code)',    'OfficialSixthForm (name)'),
        ('religious_character',  'ReligiousCharacter (code)',   'ReligiousCharacter (name)'),
        ('admissions_policy',    'AdmissionsPolicy (code)',     'AdmissionsPolicy (name)')
    ] %}
    select distinct
        '{{ field_key }}' as field,
        cast(nullif(trim("{{ code_col }}"), '') as integer) as code,
        nullif(trim("{{ name_col }}"), '') as name
    from {{ source('raw', 'gias_establishments') }}
    where nullif(trim("{{ code_col }}"), '') is not null
      and nullif(trim("{{ name_col }}"), '') is not null
    {% if not loop.last %}union all{% endif %}
    {% endfor %}
)

select r.*
from raw_pairs r
left join {{ ref('gias_code_names') }} s
    on s.field = r.field
   and s.code = r.code
   and s.name = r.name
where s.field is null
  • Step 5: Verify statically

Run:

cd /Users/tudor/projects/school_compare && \
uv run --with pyyaml python -c "import yaml; yaml.safe_load(open('pipeline/transform/models/marts/_marts_schema.yml')); print('yml OK')" && \
grep -c "_code" pipeline/transform/models/marts/dim_school.sql && \
grep -n "status_code in (1, 3)" pipeline/transform/models/marts/dim_school.sql pipeline/transform/models/marts/dim_location.sql && \
grep -rn "s\.status\b\|s\.phase\b\|s\.school_type\b\|s\.religious_character\b\|s\.admissions_policy\b\|official_sixth_form\b" pipeline/transform/models/marts/dim_school.sql | grep -v "_code"; echo "stale-name grep exit=$? (want 1)"

Expected: yml OK, both filters matched, and no stale name-column references (final grep exits 1).

  • Step 6: Commit
git add pipeline/transform/models/marts/dim_school.sql pipeline/transform/models/marts/dim_location.sql pipeline/transform/models/marts/_marts_schema.yml pipeline/transform/tests/assert_gias_code_names_match_seed.sql
git commit -m "feat(pipeline): dim_school/dim_location store GIAS codes; seed drift test

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

Task 4: Backend translates at the API boundary

Files:

  • Modify: backend/models.py (DimSchool columns)
  • Modify: backend/data_loader.py (_MAIN_QUERY + translation)
  • Test: backend/tests/test_gias_translation.py (new)

Interfaces:

  • Consumes: backend/gias_codes.py dictionaries + translate (Task 1); mart code columns (Task 3).

  • Produces: translate_gias_code_columns(df) -> df in backend/data_loader.py; after load_school_data_as_dataframe() the DataFrame carries today's name columns (phase, school_type, status, religious_denomination, admissions_policy) — every downstream consumer unchanged.

  • Step 1: Write the failing tests

Create backend/tests/test_gias_translation.py:

"""API-boundary translation: marts now carry GIAS codes; the DataFrame the
rest of the backend sees must carry today's name strings."""

import numpy as np
import pandas as pd

from backend.data_loader import translate_gias_code_columns
from backend.gias_codes import ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION


def _code_for(mapping, name):
    return next(c for c, n in mapping.items() if n == name)


def test_codes_become_todays_names():
    df = pd.DataFrame([{
        "urn": 1,
        "phase_code": float(_code_for(PHASE_OF_EDUCATION, "Primary")),
        "school_type_code": np.nan,
        "status_code": float(_code_for(ESTABLISHMENT_STATUS, "Open, but proposed to close")),
        "religious_character_code": np.nan,
        "admissions_policy_code": np.nan,
    }])
    out = translate_gias_code_columns(df)
    row = out.iloc[0]
    assert row["phase"] == "Primary"
    assert row["status"] == "Open, but proposed to close"
    assert row["school_type"] is None
    assert row["religious_denomination"] is None
    assert row["admissions_policy"] is None


def test_unknown_code_degrades_not_blanks():
    df = pd.DataFrame([{"urn": 1, "phase_code": 9999.0}])
    out = translate_gias_code_columns(df)
    assert out.iloc[0]["phase"] == "Unknown (9999)"


def test_missing_code_columns_are_a_noop():
    """Old-schema DataFrames (tests, pre-pipeline DBs) pass through untouched."""
    df = pd.DataFrame([{"urn": 1, "phase": "Primary", "status": "Open"}])
    out = translate_gias_code_columns(df)
    assert out.iloc[0]["phase"] == "Primary"
    assert out.iloc[0]["status"] == "Open"
  • Step 2: Run to verify failure

Run: cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests/test_gias_translation.py -v Expected: FAIL — ImportError: cannot import name 'translate_gias_code_columns'.

  • Step 3: Implement translation in data_loader.py

Add near the top of backend/data_loader.py (after existing imports):

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
  • Step 4: Switch _MAIN_QUERY to code columns and call the translation

In _MAIN_QUERY replace: s.phase,s.phase_code, · s.school_type,s.school_type_code, · s.religious_character AS religious_denomination,s.religious_character_code, · s.admissions_policy,s.admissions_policy_code, · s.status,s.status_code,

In load_school_data_as_dataframe(), insert the call immediately after the empty-check and before the existing normalize_school_type line:

    if df.empty:
        return df

    df = translate_gias_code_columns(df)

    # Build address string
    ...
    # Normalize school type  (existing line — now normalises the translated name)
    df["school_type"] = df["school_type"].apply(normalize_school_type)
  • Step 5: Update DimSchool in models.py

Replace phase = Column(String(100)), school_type = Column(String(100)), religious_character = Column(String(100)), admissions_policy = Column(String(50)), status = Column(String(50)) with:

    phase_code = Column(Integer)
    school_type_code = Column(Integer)
    religious_character_code = Column(Integer)
    admissions_policy_code = Column(Integer)
    status_code = Column(Integer)

Then check nothing else references the removed attributes:

grep -rn "\.phase\b\|\.school_type\b\|\.religious_character\b\|\.admissions_policy\b\|\.status\b" backend/*.py | grep -i "dimschool\|DimSchool" 

Expected: no hits (the backend reads via _MAIN_QUERY, not ORM attributes). If there are hits, update them to the _code columns + translation and note it in your report.

  • Step 6: Run the new tests and the whole backend suite

Run: cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests -v Expected: all pass — 3 new + all pre-existing (their fixtures carry name columns; translation is a no-op on them).

  • Step 7: Commit
git add backend/models.py backend/data_loader.py backend/tests/test_gias_translation.py
git commit -m "feat(api): translate GIAS codes to names at the query boundary

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

Task 5: Typesense sync translates before indexing

Files:

  • Modify: pipeline/scripts/sync_typesense.py

Interfaces:

  • Consumes: pipeline/scripts/gias_codes.py (Task 1), mart code columns (Task 3).

  • Produces: identical Typesense documents to today (facet values are names).

  • Step 1: Switch the SELECT and translate

In sync_typesense.py: add at the top (the DAG runs python scripts/sync_typesense.py, so scripts/ is sys.path[0] and a plain import works):

from gias_codes import PHASE_OF_EDUCATION, RELIGIOUS_CHARACTER, SCHOOL_TYPE, translate

In the SQL, replace s.phase,s.phase_code,, s.school_type,s.school_type_code,, s.religious_character,s.religious_character_code,.

In the document builder, replace:

        "phase": row["phase"] or "",
        "school_type": row["school_type"] or "",

with:

        "phase": translate(row["phase_code"], PHASE_OF_EDUCATION) or "",
        "school_type": translate(row["school_type_code"], SCHOOL_TYPE) or "",

and:

    if row.get("religious_character"):
        doc["religious_character"] = row["religious_character"]

with:

    religious_character = translate(row.get("religious_character_code"), RELIGIOUS_CHARACTER)
    if religious_character:
        doc["religious_character"] = religious_character
  • Step 2: Verify statically

Run:

cd /Users/tudor/projects/school_compare && \
python3 -c "import ast; ast.parse(open('pipeline/scripts/sync_typesense.py').read()); print('sync OK')" && \
grep -n "row\[\"phase\"\]\|row\[\"school_type\"\]\|row\[\"religious_character\"\]" pipeline/scripts/sync_typesense.py; echo "stale grep exit=$? (want 1)"

Expected: sync OK, no stale name-column row accesses.

  • Step 3: Commit
git add pipeline/scripts/sync_typesense.py
git commit -m "feat(pipeline): typesense sync translates GIAS codes before indexing

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

Task 6: Spec status, PR, deploy runbook

Files:

  • Modify: docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md (status line)

  • Step 1: Mark the spec implemented

Change **Status:** Approved design to **Status:** Implemented 2026-07-09 — see docs/superpowers/plans/2026-07-09-gias-code-dictionaries.md.

  • Step 2: Commit and push
git add docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md
git commit -m "docs: mark GIAS code dictionaries spec implemented

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
git push -u origin feat/gias-code-dictionaries
  • Step 3: Open the PR (Gitea API via git credential fill — token-header auth 401s)

Title: feat: GIAS classification fields stored as codes, translated in code Body must include: (1) API contract unchanged — names still served, translation at the query boundary; (2) the deploy runbook: merge → deploy → trigger school_data_daily immediately (accepted empty-API window until the marts rebuild — spec §7); (3) dictionary maintenance loop (dbt drift test warns → rerun generate_gias_codes.py → commit regenerated files); (4) no frontend/e2e changes. End with the standard generation footer.

  • Step 4: Watch CI

All PR checks must pass. Do not merge — merging triggers the deploy window; the human runs the runbook.