ss-tools/backend/src/api/routes/dashboards/_helpers.py

# #region DashboardHelpers [C:2] [TYPE Module] [SEMANTICS fastapi, dashboard, api, search, filter]
# @BRIEF Basic helper functions for dashboard route handlers — slug resolution, filter normalization.
# @LAYER Infrastructure
# @RELATION DEPENDS_ON -> [SupersetClient]
# @RELATION DEPENDS_ON -> [AsyncSupersetClient]

from typing import Any

from fastapi import HTTPException

from src.core.async_superset_client import AsyncSupersetClient
from src.core.superset_client import SupersetClient


# #region _find_dashboard_id_by_slug [C:2] [TYPE Function]
# @BRIEF Resolve dashboard numeric ID by slug using Superset list endpoint.
# @PRE  `dashboard_slug` is non-empty.
# @POST  Returns dashboard ID when found, otherwise None.
def _find_dashboard_id_by_slug(
    client: SupersetClient,
    dashboard_slug: str,
) -> int | None:
    query_variants = [
        {
            "filters": [{"col": "slug", "opr": "eq", "value": dashboard_slug}],
            "page": 0,
            "page_size": 1,
        },
        {
            "filters": [{"col": "slug", "op": "eq", "value": dashboard_slug}],
            "page": 0,
            "page_size": 1,
        },
    ]

    for query in query_variants:
        try:
            _count, dashboards = client.get_dashboards_page(query=query)
            if dashboards:
                resolved_id = dashboards[0].get("id")
                if resolved_id is not None:
                    return int(resolved_id)
        except Exception:
            continue

    return None


# #endregion _find_dashboard_id_by_slug


# #region _resolve_dashboard_id_from_ref [C:2] [TYPE Function]
# @BRIEF Resolve dashboard ID from slug-first reference with numeric fallback.
# @PRE  `dashboard_ref` is provided in route path.
# @POST  Returns a valid dashboard ID or raises HTTPException(404).
def _resolve_dashboard_id_from_ref(
    dashboard_ref: str,
    client: SupersetClient,
) -> int:
    normalized_ref = str(dashboard_ref or "").strip()
    if not normalized_ref:
        raise HTTPException(status_code=404, detail="Dashboard not found")

    # Slug-first: even if ref looks numeric, try slug first.
    slug_match_id = _find_dashboard_id_by_slug(client, normalized_ref)
    if slug_match_id is not None:
        return slug_match_id

    if normalized_ref.isdigit():
        return int(normalized_ref)

    raise HTTPException(status_code=404, detail="Dashboard not found")


# #endregion _resolve_dashboard_id_from_ref


# #region _find_dashboard_id_by_slug_async [C:2] [TYPE Function]
# @BRIEF Resolve dashboard numeric ID by slug using async Superset list endpoint.
# @PRE  dashboard_slug is non-empty.
# @POST  Returns dashboard ID when found, otherwise None.
async def _find_dashboard_id_by_slug_async(
    client: AsyncSupersetClient,
    dashboard_slug: str,
) -> int | None:
    query_variants = [
        {
            "filters": [{"col": "slug", "opr": "eq", "value": dashboard_slug}],
            "page": 0,
            "page_size": 1,
        },
        {
            "filters": [{"col": "slug", "op": "eq", "value": dashboard_slug}],
            "page": 0,
            "page_size": 1,
        },
    ]

    for query in query_variants:
        try:
            _count, dashboards = await client.get_dashboards_page(query=query)
            if dashboards:
                resolved_id = dashboards[0].get("id")
                if resolved_id is not None:
                    return int(resolved_id)
        except Exception:
            continue

    return None


# #endregion _find_dashboard_id_by_slug_async


# #region _resolve_dashboard_id_from_ref_async [C:2] [TYPE Function]
# @BRIEF Resolve dashboard ID from slug-first reference using async Superset client.
# @PRE  dashboard_ref is provided in route path.
# @POST  Returns valid dashboard ID or raises HTTPException(404).
async def _resolve_dashboard_id_from_ref_async(
    dashboard_ref: str,
    client: AsyncSupersetClient,
) -> int:
    normalized_ref = str(dashboard_ref or "").strip()
    if not normalized_ref:
        raise HTTPException(status_code=404, detail="Dashboard not found")

    slug_match_id = await _find_dashboard_id_by_slug_async(client, normalized_ref)
    if slug_match_id is not None:
        return slug_match_id

    if normalized_ref.isdigit():
        return int(normalized_ref)

    raise HTTPException(status_code=404, detail="Dashboard not found")


# #endregion _resolve_dashboard_id_from_ref_async


# #region _normalize_filter_values [C:2] [TYPE Function]
# @BRIEF Normalize query filter values to lower-cased non-empty tokens.
# @PRE  values may be None or list of strings.
# @POST  Returns trimmed normalized list preserving input order.
def _normalize_filter_values(values: list[str] | None) -> list[str]:
    if not values:
        return []
    normalized: list[str] = []
    for value in values:
        token = str(value or "").strip().lower()
        if token:
            normalized.append(token)
    return normalized


# #endregion _normalize_filter_values


# #region _dashboard_git_filter_value [C:2] [TYPE Function]
# @BRIEF Build comparable git status token for dashboards filtering.
# @PRE  dashboard payload may contain git_status or None.
# @POST  Returns one of ok|diff|no_repo|error|pending.
def _dashboard_git_filter_value(dashboard: dict[str, Any]) -> str:
    git_status = dashboard.get("git_status") or {}
    sync_status = str(git_status.get("sync_status") or "").strip().upper()
    has_repo = git_status.get("has_repo")
    if has_repo is False or sync_status == "NO_REPO":
        return "no_repo"
    if sync_status == "DIFF":
        return "diff"
    if sync_status == "OK":
        return "ok"
    if sync_status == "ERROR":
        return "error"
    return "pending"


# #endregion _dashboard_git_filter_value


# #endregion DashboardHelpers