SPB Git

spb/qwhpi Public

QHPI — Quebec Housing Price Index: quality-adjusted, hierarchically pooled housing price indexes.

Python 63.9% TypeScript 25.4% CSS 5.5% TeX 3.5% SQL 0.8% Makefile 0.5% Dockerfile 0.5%
2.3 KB · 64 lines python
Raw Blame History
1# =============================================================================2# QWHPI — Quebec Weekly Housing Price Index3# Author  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# File    : api/app/routers/geographies.py6# Purpose : Geography hierarchy + coverage matrix endpoints.7# =============================================================================8"""Geography and coverage endpoints."""910from __future__ import annotations1112import polars as pl13from fastapi import APIRouter, Query, Request, Response1415from app.deps import csv_response, etag_for16from app.schemas import GeographiesResponse, GeographyNode17from app.services import data1819router = APIRouter()202122@router.get("/geographies", response_model=GeographiesResponse)23def geographies(request: Request, response: Response):24    """Published series hierarchy (province → regions → municipalities)."""25    if etag_for(request, response):26        return Response(status_code=304)27    c = data.canonical()28    nodes = (29        c.group_by(["geography_id", "geography_name", "geography_level"])30        .agg(pl.col("property_type").unique().sort().alias("types"))31        .sort(["geography_level", "geography_id"])32    )33    return GeographiesResponse(34        data_vintage=data.data_vintage(),35        published_series=[36            GeographyNode(geography_id=r["geography_id"],37                          geography_name=r["geography_name"],38                          geography_level=r["geography_level"],39                          property_types=r["types"])40            for r in nodes.iter_rows(named=True)41        ],42    )434445@router.get("/geographies/coverage")46def coverage_matrix(47    request: Request,48    response: Response,49    status: str | None = Query(None, pattern="^(published|conditional|not_published)$"),50    format: str = Query("json", pattern="^(json|csv)$"),51):52    """Full municipality × type coverage matrix with publication status."""53    df = data.coverage()54    if status:55        df = df.filter(pl.col("coverage_status") == status)56    if format == "csv":57        return csv_response(df)58    if etag_for(request, response, str(status)):59        return Response(status_code=304)60    return {61        "coverage": df.rename({"propertyType": "property_type"}).to_dicts(),62        "author": "Simon-Pierre Boucher", "contact": "contact@spboucher.ai",63    }64