pipeline_miami/verify/connectors/served_db.py
Served verbatim from the project repository. Internal working document conventions apply: documents may reference file paths, branch names, and findings-ledger anchors from the repo.
"""Served-layer reader for Miami metrics. Read-only Postgres."""
from __future__ import annotations
from pipeline_miami.verify import common
_SQL = """
SELECT m.school_number, m.year, m.value::float8, m.suppressed, m.denominator::float8,
s.school_type::text, s.is_charter, s.activity_code, s.school_level
FROM miami_school_year_metrics m
JOIN miami_schools s ON s.school_number = m.school_number
WHERE m.metric_key = %s AND m.subgroup::text = %s
"""
def fetch(metric_key: str, subgroup: str = "ALL") -> dict[tuple[str, str], dict]:
import psycopg
out: dict[tuple[str, str], dict] = {}
with psycopg.connect(common.served_db_dsn()) as conn:
conn.read_only = True
with conn.cursor() as cur:
cur.execute(_SQL, (metric_key, subgroup))
for num, year, value, suppressed, n, stype, charter, act, level in cur:
out[(str(num), 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, "school_type": stype, "is_charter": charter,
"activity_code": act, "school_level": level,
}
return out
def directory() -> dict[str, dict]:
import psycopg
out: dict[str, dict] = {}
with psycopg.connect(common.served_db_dsn()) as conn:
conn.read_only = True
with conn.cursor() as cur:
cur.execute("SELECT school_number, name, school_type::text, is_charter, activity_code, school_level, include_in_default_comparisons, fish_capacity_permanent FROM miami_schools")
for num, name, stype, charter, act, level, inc, cap in cur:
out[str(num)] = {"name": name, "school_type": stype, "is_charter": charter, "activity_code": act, "school_level": level, "include": inc, "fish_capacity_permanent": cap}
return out