pipeline_philly/verify/connectors/served_db.py

"""Served-layer reader for Philly metrics. Read-only Postgres."""
from __future__ import annotations

from pipeline_philly.verify import common

_SQL_PHILLY = """
    SELECT m.school_ulcs,
           m.year,
           m.value::float8 AS value,
           m.suppressed,
           m.denominator::float8 AS n,
           m.subgroup::text AS subgroup,
           m.population_cut,
           s.school_type::text AS school_type,
           s.learning_network
      FROM philly_school_year_metrics m
      JOIN philly_schools s ON s.ulcs_code = m.school_ulcs
     WHERE m.metric_key = %s
       AND m.subgroup::text = %s
       AND m.population_cut = %s
"""


def fetch(
    metric_key: str,
    subgroup: str = "ALL",
    population_cut: str = "n/a",
) -> dict[tuple[str, str], dict]:
    """Return {(ulcs, year): {value, suppressed, n, subgroup, ...}} for one metric × subgroup × cut."""
    dsn = common.served_db_dsn()
    try:
        import psycopg
    except ImportError as e:
        raise SystemExit("psycopg not installed — pip install 'psycopg[binary]'") from e

    out: dict[tuple[str, str], dict] = {}
    with psycopg.connect(dsn) as conn:
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute(_SQL_PHILLY, (metric_key, subgroup, population_cut))
            for ulcs, year, value, suppressed, n, sg, cut, stype, network in cur:
                out[(str(ulcs), str(year))] = {
                    "value": float(value) if value is not None else None,
                    "suppressed": bool(suppressed),
                    "n": float(n) if n is not None else None,
                    "subgroup": sg,
                    "population_cut": cut,
                    "school_type": stype,
                    "learning_network": network,
                }
    return out