Webapp Development

Key Rules:

• apply_page_config() MUST be the first Streamlit command
• Use fixed header navigation (NOT sidebar)
• Call render_header() with the current page name

Step 2: Build Components

Extract reusable UI elements into src/components/:

# src/components/kpi_cards.py
import streamlit as st

def render_kpi_row(kpis: list[dict]):
    """Render a row of KPI cards with brand theming."""
    cols = st.columns(len(kpis))
    for col, kpi in zip(cols, kpis):
        with col:
            st.metric(
                label=kpi["label"],
                value=kpi["value"],
                delta=kpi.get("delta"),
            )

Step 3: Implement Service Layer

# src/services/record_service.py
from loguru import logger
import pandas as pd

class RecordService:
    def __init__(self, repository):
        self._repo = repository

    def get_dashboard_data(self, filter_type: str, date_str: str = "") -> dict:
        """Get all data needed for the dashboard page."""
        logger.info(f"Loading dashboard data: filter={filter_type}, date={date_str}")
        return {
            "kpis": self._repo.get_kpi_summary(filter_type, date_str),
            "trend": self._repo.get_trend_data(filter_type, date_str),
            "distribution": self._repo.get_workbasket_distribution(filter_type, date_str),
        }

Step 4: Implement Repository Pattern

# src/services/record_repository.py
import pandas as pd
from src.ingestion_config.data_sources import get_data_source_config

class RecordRepository:
    def __init__(self):
        self._config = get_data_source_config()
        self._adapter = self._create_adapter()

    def execute_query(self, query_name: str, **params) -> pd.DataFrame:
        """Execute a named query from queries.yaml."""
        query_config = self._config.get_query(query_name)
        sql = query_config["sql"].format(table=self._config.get_table(query_config["entity"]))
        return self._adapter.fetch_dataframe(sql, params.get("params", []))

Step 5: Externalize SQL Queries

All SQL goes in src/ingestion_config/queries.yaml:

home_charts:
  workbasket_distribution:
    description: "Count open cases by workbasket for dashboard"
    entity: pega_cases
    sql: |
      SELECT
          ISNULL(WorkbasketHeading, 'Unknown') AS workbasket_heading,
          COUNT(*) AS count
      FROM {table}
      WHERE [Created Date Time] >= ?
        AND [Created Date Time] < ?
        AND Status = 'Open'
      GROUP BY WorkbasketHeading
      ORDER BY count DESC

Step 6: Apply Brand Theming

# Use brand color constants
BRAND_RED = "#E10A0A"
BRAND_NAVY = "#1F2937"
LIGHT_BG = "#F8F9FA"

CHART_COLORS = [BRAND_RED, BRAND_NAVY, "#6B7280", "#9CA3AF", "#D1D5DB"]

# Load CSS theme
def load_theme():
    css_path = Path(__file__).parent.parent / "styles" / "theme.css"
    if css_path.exists():
        st.markdown(f"<style>{css_path.read_text()}</style>", unsafe_allow_html=True)

Step 7: Manage Session State

def init_session_state():
    defaults = {
        "authenticated": False,
        "current_user": None,
        "selected_filters": {},
        "page_number": 1,
        "filter_type": "daily",
    }
    for key, value in defaults.items():
        if key not in st.session_state:
            st.session_state[key] = value

def on_filter_change():
    """Reset pagination when filters change."""
    st.session_state.page_number = 1

Step 8: Add Caching

from datetime import timedelta

@st.cache_data(ttl=timedelta(minutes=15), show_spinner="Loading data...")
def load_dashboard_data(filter_type: str, date_str: str) -> dict:
    service = RecordService(get_repo())
    return service.get_dashboard_data(filter_type, date_str)

@st.cache_resource
def get_repo():
    return RecordRepository()

Step 9: Error Handling

from loguru import logger