""" SQLAlchemy models — all tables live in the marts schema, built by dbt. Read-only: the pipeline writes to these tables; the backend only reads. """ from sqlalchemy import Column, Integer, String, Float, Boolean, Date, Text, Index from .database import Base MARTS = {"schema": "marts"} class DimSchool(Base): """Canonical school dimension — one row per active URN.""" __tablename__ = "dim_school" __table_args__ = MARTS urn = Column(Integer, primary_key=True) school_name = Column(String(255), nullable=False) phase_code = Column(Integer) school_type_code = Column(Integer) academy_trust_name = Column(String(255)) academy_trust_uid = Column(String(20)) religious_character_code = Column(Integer) gender = Column(String(20)) age_range = Column(String(20)) has_sixth_form = Column(Boolean) capacity = Column(Integer) total_pupils = Column(Integer) headteacher_name = Column(String(200)) website = Column(String(255)) telephone = Column(String(30)) status_code = Column(Integer) nursery_provision = Column(Boolean) admissions_policy_code = Column(Integer) # Denormalised Ofsted summary (updated by monthly pipeline) ofsted_grade = Column(Integer) ofsted_date = Column(Date) ofsted_framework = Column(String(20)) class DimLocation(Base): """School location — address, lat/lng from easting/northing (BNG→WGS84).""" __tablename__ = "dim_location" __table_args__ = MARTS urn = Column(Integer, primary_key=True) address_line1 = Column(String(255)) address_line2 = Column(String(255)) town = Column(String(100)) county = Column(String(100)) postcode = Column(String(20)) local_authority_code = Column(Integer) local_authority_name = Column(String(100)) parliamentary_constituency = Column(String(100)) urban_rural = Column(String(50)) easting = Column(Integer) northing = Column(Integer) latitude = Column(Float) longitude = Column(Float) # geom is a PostGIS geometry — not mapped to SQLAlchemy (accessed via raw SQL) class KS2Performance(Base): """KS2 attainment — one row per URN per year (includes predecessor data).""" __tablename__ = "fact_ks2_performance" __table_args__ = ( Index("ix_ks2_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) source_urn = Column(Integer) total_pupils = Column(Integer) eligible_pupils = Column(Integer) # Core attainment rwm_expected_pct = Column(Float) rwm_high_pct = Column(Float) reading_expected_pct = Column(Float) reading_high_pct = Column(Float) reading_avg_score = Column(Float) reading_progress = Column(Float) writing_expected_pct = Column(Float) writing_high_pct = Column(Float) writing_progress = Column(Float) maths_expected_pct = Column(Float) maths_high_pct = Column(Float) maths_avg_score = Column(Float) maths_progress = Column(Float) # Progress confidence intervals + writing working-towards (published # for years with progress measures, i.e. up to 2022/23) reading_progress_lower_ci = Column(Float) reading_progress_upper_ci = Column(Float) writing_progress_lower_ci = Column(Float) writing_progress_upper_ci = Column(Float) writing_working_towards_pct = Column(Float) maths_progress_lower_ci = Column(Float) maths_progress_upper_ci = Column(Float) gps_expected_pct = Column(Float) gps_high_pct = Column(Float) gps_avg_score = Column(Float) science_expected_pct = Column(Float) # Absence reading_absence_pct = Column(Float) writing_absence_pct = Column(Float) maths_absence_pct = Column(Float) gps_absence_pct = Column(Float) science_absence_pct = Column(Float) # Gender rwm_expected_boys_pct = Column(Float) rwm_high_boys_pct = Column(Float) rwm_expected_girls_pct = Column(Float) rwm_high_girls_pct = Column(Float) # Disadvantaged rwm_expected_disadvantaged_pct = Column(Float) rwm_expected_non_disadvantaged_pct = Column(Float) disadvantaged_gap = Column(Float) # Context disadvantaged_pct = Column(Float) eal_pct = Column(Float) sen_support_pct = Column(Float) sen_ehcp_pct = Column(Float) stability_pct = Column(Float) class FactOfstedInspection(Base): """Full Ofsted inspection history — one row per inspection.""" __tablename__ = "fact_ofsted_inspection" __table_args__ = ( Index("ix_ofsted_urn_date", "urn", "inspection_date"), MARTS, ) urn = Column(Integer, primary_key=True) inspection_date = Column(Date, primary_key=True) inspection_type = Column(String(100)) framework = Column(String(20)) overall_effectiveness = Column(Integer) quality_of_education = Column(Integer) behaviour_attitudes = Column(Integer) personal_development = Column(Integer) leadership_management = Column(Integer) early_years_provision = Column(Integer) sixth_form_provision = Column(Integer) # Ungraded (Section 8) inspection: raw outcome text and the grade parsed from # it (fallback for schools with no graded overall effectiveness). ungraded_outcome = Column(String(100)) ungraded_grade = Column(Integer) rc_safeguarding_met = Column(Boolean) rc_inclusion = Column(Integer) rc_curriculum_teaching = Column(Integer) rc_achievement = Column(Integer) rc_attendance_behaviour = Column(Integer) rc_personal_development = Column(Integer) rc_leadership_governance = Column(Integer) rc_early_years = Column(Integer) rc_sixth_form = Column(Integer) # Start date of the report-card inspection itself (renewed framework, # Nov 2025+). Null for rows without report-card grades. rc_inspection_date = Column(Date) report_url = Column(Text) class FactAdmissions(Base): """School admissions — one row per URN per year.""" __tablename__ = "fact_admissions" __table_args__ = ( Index("ix_admissions_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) school_phase = Column(String(50)) places_offered = Column(Integer) total_applications = Column(Integer) first_preference_applications = Column(Integer) first_preference_offers = Column(Integer) total_offers = Column(Integer) second_preference_offers = Column(Integer) third_preference_offers = Column(Integer) cross_la_applications = Column(Integer) cross_la_offers = Column(Integer) first_preference_offer_pct = Column(Float) oversubscription_ratio = Column(Float) oversubscribed = Column(Boolean) admissions_policy = Column(String(100)) class FactAdmissionDistance(Base): """Last distance offered — one row per URN per year. Separate from FactAdmissions because the source is separate: EES publishes admissions for the whole country, whereas cut-off distances exist only for the local authorities that choose to publish them (57 at the time of writing), and the two refresh on unrelated timetables. """ __tablename__ = "fact_admission_distance" __table_args__ = ( Index("ix_admission_distance_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) # Straight-line distance in metres from the school to the last home offered # a place that year. distance_m = Column(Float) # How many admission routes (ability bands, separate reception/junior # intakes) were collapsed into distance_m. >1 means the figure is the # furthest of several and the page must say so. route_count = Column(Integer) la_code = Column(Integer) la_name = Column(String(100)) # The unit the council published in, so the page can lead with the unit a # parent was given rather than always converting. distance_unit_raw = Column(String(20)) source_file = Column(Text) class FactPupilCharacteristics(Base): """School pupil composition from EES census — one row per URN per year.""" __tablename__ = "fact_pupil_characteristics" __table_args__ = ( Index("ix_pupil_chars_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) phase_type_grouping = Column(String(50)) total_pupils = Column(Integer) female_pupils = Column(Integer) male_pupils = Column(Integer) fsm_pct = Column(Float) eal_pct = Column(Float) class FactDeprivation(Base): """IDACI deprivation index — one row per URN.""" __tablename__ = "fact_deprivation" __table_args__ = MARTS urn = Column(Integer, primary_key=True) lsoa_code = Column(String(20)) idaci_score = Column(Float) idaci_decile = Column(Integer) class FactFinance(Base): """FBIT financial benchmarking — one row per URN per year.""" __tablename__ = "fact_finance" __table_args__ = ( Index("ix_finance_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) per_pupil_spend = Column(Float) staff_cost_pct = Column(Float) teacher_cost_pct = Column(Float) support_staff_cost_pct = Column(Float) premises_cost_pct = Column(Float) class CensusBenchmark(Base): """State-school context benchmarks from the pupil census — one row per phase. fsm_pct / eal_pct are pupil-weighted means. Computed at import time; consumers label them "state-school average (computed from our dataset)". """ __tablename__ = "fact_census_benchmarks" __table_args__ = MARTS phase = Column(String(20), primary_key=True) year = Column(Integer) fsm_pct = Column(Float) eal_pct = Column(Float) median_pupils = Column(Integer) class Ks4NationalAverage(Base): """Official DfE KS4 national headline averages — one row per academic year. gcse_grade_91_pct has no official national series and is always NULL. """ __tablename__ = "fact_ks4_national_averages" __table_args__ = MARTS year = Column(Integer, primary_key=True) attainment_8_score = Column(Float) progress_8_score = Column(Float) english_maths_standard_pass_pct = Column(Float) english_maths_strong_pass_pct = Column(Float) ebacc_entry_pct = Column(Float) ebacc_standard_pass_pct = Column(Float) ebacc_strong_pass_pct = Column(Float) ebacc_avg_score = Column(Float) gcse_grade_91_pct = Column(Float) class Ks2NationalAverage(Base): """Official DfE KS2 national headline averages — one row per academic year.""" __tablename__ = "fact_ks2_national_averages" __table_args__ = MARTS year = Column(Integer, primary_key=True) rwm_expected_pct = Column(Float) rwm_high_pct = Column(Float) reading_expected_pct = Column(Float) reading_high_pct = Column(Float) reading_avg_score = Column(Float) writing_expected_pct = Column(Float) writing_gd_pct = Column(Float) maths_expected_pct = Column(Float) maths_high_pct = Column(Float) maths_avg_score = Column(Float) gps_expected_pct = Column(Float) gps_high_pct = Column(Float) gps_avg_score = Column(Float) science_expected_pct = Column(Float) class FactKs4Destinations(Base): """KS4 leavers destinations — one row per URN, year, pupil group, measure. Long format rather than wide because pupil_group is a real third dimension. `status` is load-bearing: 'suppressed' means DfE withheld a figure it considered disclosive and the page must print "withheld"; 'not_applicable' means the measure does not apply and the page must print nothing. `pupils` is null for both, so collapsing status to a null check loses the difference — and the categories sum to the cohort, so a consumer that treats a withheld cell as zero republishes what DfE hid. """ __tablename__ = "fact_ks4_destinations" __table_args__ = ( Index("ix_ks4_dest_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) pupil_group = Column(String(20), primary_key=True) destination_measure = Column(String(40), primary_key=True) cohort_pupils = Column(Integer) pupils = Column(Integer) percentage = Column(Float) status = Column(String(20)) class FactKs5Destinations(Base): """16-18 study leavers destinations — same grain as FactKs4Destinations.""" __tablename__ = "fact_ks5_destinations" __table_args__ = ( Index("ix_ks5_dest_urn_year", "urn", "year"), MARTS, ) urn = Column(Integer, primary_key=True) year = Column(Integer, primary_key=True) pupil_group = Column(String(20), primary_key=True) destination_measure = Column(String(40), primary_key=True) cohort_pupils = Column(Integer) pupils = Column(Integer) percentage = Column(Float) status = Column(String(20))