nprasad2077
11a5cc0f18
CI Health Check & Quality Gate / Code, dbt & Live App Health Check (pull_request) Successful in 1m11s
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
from pathlib import Path
|
|
import duckdb
|
|
import pandas as pd
|
|
import streamlit as st
|
|
|
|
|
|
def get_db_path() -> Path:
|
|
"""Resolve local DuckDB file path across reports and data/DB locations."""
|
|
repo_root = Path(__file__).resolve().parent.parent.parent
|
|
candidates = [
|
|
repo_root / "dbt_nba" / "reports" / "sources" / "nba" / "dbt_nba.duckdb",
|
|
repo_root / "data" / "DB" / "dbt_nba.duckdb",
|
|
Path("dbt_nba/reports/sources/nba/dbt_nba.duckdb"),
|
|
Path("data/DB/dbt_nba.duckdb"),
|
|
]
|
|
for path in candidates:
|
|
if path.exists():
|
|
return path.resolve()
|
|
return candidates[0]
|
|
|
|
|
|
@st.cache_resource
|
|
def get_connection():
|
|
"""Persistent read-only DuckDB connection."""
|
|
db_path = get_db_path()
|
|
if not db_path.exists():
|
|
st.error(
|
|
f"Database file not found at `{db_path}`. "
|
|
"Run `./scripts/run_app.sh --sync` or `./scripts/pipeline.sh` to initialize."
|
|
)
|
|
st.stop()
|
|
return duckdb.connect(str(db_path), read_only=True)
|
|
|
|
|
|
@st.cache_data(ttl=3600)
|
|
def query(sql: str, params: list | None = None) -> pd.DataFrame:
|
|
"""Execute SQL (optionally parameterized) and return cached DataFrame."""
|
|
conn = get_connection()
|
|
if params:
|
|
return conn.execute(sql, params).df()
|
|
return conn.execute(sql).df()
|
|
|
|
|
|
@st.cache_data(ttl=3600)
|
|
def get_seasons() -> pd.DataFrame:
|
|
"""Return available seasons ordered most recent first."""
|
|
return query(
|
|
"SELECT season_start_year, season_display "
|
|
"FROM main_marts.dim_seasons "
|
|
"ORDER BY season_start_year DESC"
|
|
)
|
|
|
|
|
|
def get_selected_season() -> tuple[int, str]:
|
|
"""Return (season_start_year, season_display) from session state."""
|
|
seasons = get_seasons()
|
|
season_options = seasons["season_start_year"].tolist()
|
|
season_labels = dict(zip(seasons["season_start_year"], seasons["season_display"]))
|
|
if "season" not in st.session_state or st.session_state["season"] not in season_labels:
|
|
st.session_state["season"] = season_options[0]
|
|
season = st.session_state["season"]
|
|
return season, season_labels[season]
|