Get Data Validation Results with GraphQL

SPECKLE_HOST=https://app.speckle.systems
SPECKLE_TOKEN=your_personal_access_token
SPECKLE_PROJECT_ID=your_project_id

%pip install -q specklepy pandas python-dotenv

import os
from collections import defaultdict

import pandas as pd
from dotenv import load_dotenv
from gql import gql
from specklepy.api.client import SpeckleClient

load_dotenv()

HOST = os.getenv("SPECKLE_HOST", "https://app.speckle.systems")
TOKEN = os.getenv("SPECKLE_TOKEN")
PROJECT_ID = os.getenv("SPECKLE_PROJECT_ID")

if not TOKEN:
    raise ValueError("Set SPECKLE_TOKEN in your environment.")
if not PROJECT_ID:
    raise ValueError("Set SPECKLE_PROJECT_ID in your environment.")

client = SpeckleClient(host=HOST)
client.authenticate_with_token(TOKEN)

DEFAULT_DISPLAY_CONFIG = {
    "passThreshold": 0.9,
    "warningThreshold": None,
    "rulePassThreshold": {},
    "ruleWarningThreshold": {},
    "ruleSeverity": {},
}

LIST_CHECKS_QUERY = gql("""
query ProjectValidationChecks($projectId: String!) {
  projectInsights(projectId: $projectId, type: "model_validation") {
    id
    name
    metadata
    aggregateResults(limit: 1) {
      timestamp
      summary
    }
  }
}
""")


def run_query(query, variables: dict | None = None) -> dict:
    if variables:
        return client.httpclient.execute(query, variable_values=variables)
    return client.execute_query(query)


def resolve_thresholds(display_config: dict, rule_name: str | None = None) -> dict:
    cfg = {**DEFAULT_DISPLAY_CONFIG, **(display_config or {})}
    if rule_name:
        pass_t = cfg["rulePassThreshold"].get(rule_name, cfg["passThreshold"])
        warn_map = cfg.get("ruleWarningThreshold") or {}
        warn_t = warn_map[rule_name] if rule_name in warn_map else cfg.get("warningThreshold")
        severity = cfg.get("ruleSeverity", {}).get(rule_name, "error")
        return {"passThreshold": pass_t, "warningThreshold": warn_t, "severity": severity}
    return {
        "passThreshold": cfg["passThreshold"],
        "warningThreshold": cfg.get("warningThreshold"),
        "severity": "error",
    }


def compute_pass_rate(summary: dict) -> float | None:
    pass_n = summary.get("pass", 0) or 0
    fail_n = summary.get("fail", 0) or 0
    total = pass_n + fail_n
    if total == 0:
        return None
    return pass_n / total


def compute_score_pct(summary: dict) -> int | None:
    rate = compute_pass_rate(summary)
    if rate is None:
        return None
    return round(rate * 100)


def compute_status(
    pass_rate: float | None,
    display_config: dict,
    rule_name: str | None = None,
) -> str:
    if pass_rate is None:
        return "na"
    thresholds = resolve_thresholds(display_config, rule_name)
    if rule_name and thresholds.get("severity") == "info":
        return "info"
    pass_t = thresholds["passThreshold"]
    warn_t = thresholds.get("warningThreshold")
    if pass_rate >= pass_t:
        return "pass"
    if warn_t is not None and pass_rate >= warn_t:
        return "warning"
    return "fail"


def checks_to_kpi_df(checks: list[dict]) -> pd.DataFrame:
    rows = []
    for check in checks:
        agg_list = check.get("aggregateResults") or []
        agg = agg_list[0] if agg_list else None
        summary = (agg or {}).get("summary") or {}
        metadata = check.get("metadata") or {}
        display_config = metadata.get("displayConfig") or {}
        pass_rate = compute_pass_rate(summary)
        rows.append(
            {
                "name": check.get("name"),
                "insight_id": check.get("id"),
                "pass": summary.get("pass", 0),
                "fail": summary.get("fail", 0),
                "score_pct": compute_score_pct(summary),
                "status": compute_status(pass_rate, display_config),
                "evaluated_at": (agg or {}).get("timestamp"),
            }
        )
    return pd.DataFrame(rows)

result = run_query(LIST_CHECKS_QUERY, {"projectId": PROJECT_ID})
checks = result.get("projectInsights") or []
kpi_df = checks_to_kpi_df(checks)
kpi_df

def query_result_to_rules_df(result: dict, display_config: dict) -> pd.DataFrame:
    rule_entries: dict[str, list[dict]] = defaultdict(list)
    for row in result.get("rows", []):
        vals = row.get("values", {})
        rule_name = vals.get("rule")
        status = vals.get("status")
        if rule_name == "_total" or status not in ("pass", "fail"):
            continue
        rule_entries[rule_name].append(vals)

    out = []
    for rule_name, entries in rule_entries.items():
        max_gate = max(int(e.get("gate", 0)) for e in entries)
        pass_n = sum(
            int(e.get("count", 0))
            for e in entries
            if e.get("status") == "pass" and int(e.get("gate", 0)) == max_gate
        )
        fail_n = sum(
            int(e.get("count", 0))
            for e in entries
            if e.get("status") == "fail" and int(e.get("gate", 0)) == max_gate
        )
        total = pass_n + fail_n
        ratio = pass_n / total if total else 0.0
        out.append(
            {
                "rule": rule_name,
                "pass": pass_n,
                "fail": fail_n,
                "ratio": ratio,
                "status": compute_status(ratio if total else None, display_config, rule_name),
            }
        )
    return pd.DataFrame(out)


def aggregate_history_to_df(aggregate_results: list[dict], display_config: dict) -> pd.DataFrame:
    rows = []
    for entry in reversed(aggregate_results):
        summary = entry.get("summary") or {}
        rate = compute_pass_rate(summary)
        rows.append(
            {
                "timestamp": entry.get("timestamp"),
                "pass": summary.get("pass", 0),
                "fail": summary.get("fail", 0),
                "score_pct": compute_score_pct(summary),
                "status": compute_status(rate, display_config),
            }
        )
    return pd.DataFrame(rows)

import time
from gql import gql
from specklepy.api.client import SpeckleClient

WAIT_FOR_AGGREGATE_QUERY = gql("""
query WaitForAggregate($projectId: String!, $insightId: String!) {
  insight(id: $insightId, projectId: $projectId) {
    aggregateResults(limit: 1) {
      timestamp
      summary
    }
  }
}
""")


def wait_for_aggregate_result(
    client: SpeckleClient,
    project_id: str,
    insight_id: str,
    interval_sec: float = 15.0,
    max_attempts: int = 40,
) -> dict | None:
    variables = {"projectId": project_id, "insightId": insight_id}
    for _ in range(max_attempts):
        result = client.httpclient.execute(WAIT_FOR_AGGREGATE_QUERY, variable_values=variables)
        results = (result.get("insight") or {}).get("aggregateResults") or []
        if results:
            return results[0]
        time.sleep(interval_sec)
    return None