sql_qa_pair/app/operators/get_dashboard_info.py

"""Operator to assemble dashboard metadata from DuckDB for downstream prompts."""

from __future__ import annotations

import json
from typing import Any, Dict, List

from app.data_access import DuckDBClient
from app.endpoints.schema import CardFilter, CardInfo, DashboardInfo


def _build_filters(filter_rows: List[Dict[str, Any]]) -> Dict[str, List[CardFilter]]:
    filters_by_card: Dict[str, List[CardFilter]] = {}
    for row in filter_rows:
        options_raw = row.get("options") or "[]"
        try:
            options = json.loads(options_raw)
        except Exception:
            options = []
        card_filter = CardFilter(
            card_id=row["card_id"],
            filter_id=row["filter_id"],
            type=row.get("filter_type", "F"),
            where_clause=row.get("where_clause", ""),
            default_value=row.get("default_value"),
            options=options,
        )
        filters_by_card.setdefault(row["card_id"], []).append(card_filter)
    return filters_by_card


def get_dashboard_info(input_args: Dict[str, Any]) -> DashboardInfo:
    """Retrieve dashboard info + card definitions + filters for given ids."""

    dashboard_id = input_args.get("dashboard_id")
    card_ids = input_args.get("card_ids") or []
    bbk_id = input_args.get("bbk")
    if not dashboard_id:
        raise ValueError("dashboard_id is required in input_args")

    client = DuckDBClient()
    mapping_rows = client.fetch_dashboard_cards(dashboard_id, card_ids, bbk_id)
    if not mapping_rows:
        raise ValueError(f"Dashboard {dashboard_id} not found or no cards match")

    target_card_ids = [row["card_id"] for row in mapping_rows]
    card_defs = {row["card_id"]: row for row in client.fetch_card_definition(target_card_ids)}
    filter_rows = client.fetch_card_filters(target_card_ids)
    filters_by_card = _build_filters(filter_rows)

    dashboard_meta = client.fetch_dashboard_info(dashboard_id) or {
        "dashboard_id": dashboard_id,
        "dashboard_name": mapping_rows[0].get("dashboard_name", ""),
        "dashboard_desc": "",
        "folder_path": "",
    }

    cards: List[CardInfo] = []
    for row in mapping_rows:
        card_id = row["card_id"]
        card_def = card_defs.get(card_id, {})
        cards.append(
            CardInfo(
                card_id=card_id,
                card_name=card_def.get("card_name", row.get("card_name", "")),
                card_desc=card_def.get("card_desc", ""),
                dataset_id=row.get("dataset_id", ""),
                sql_select=card_def.get("sql_select", ""),
                sql_where={},
                sql_groupby=card_def.get("sql_groupby", ""),
                filters=filters_by_card.get(card_id, []),
            )
        )

    return DashboardInfo(
        dashboard_id=dashboard_meta.get("dashboard_id", dashboard_id),
        dashboard_name=dashboard_meta.get("dashboard_name", ""),
        dashboard_desc=dashboard_meta.get("dashboard_desc", ""),
        folder_path=dashboard_meta.get("folder_path", ""),
        cards=cards,
    )