[Python] Script for reading data from Grafana panels

Discussão em 'Python' iniciado por Stack, Agosto 6, 2025.

  1. Stack

    Stack Membro Participativo

    I wanted your help. I'm creating a script to read data from a panel created in a Grafana dashboard. It would bring the data. If there's any N/A or No data return, it would bring that result in a CSV file. If there is data, it would return OK. However, it brings everything except what I want. I'd like some guidance. I'll send the script below.

    import logging
    import requests
    import csv
    from pathlib import Path
    from datetime import datetime

    # Caminho para o arquivo de credenciais
    CREDENTIALS_PATH = Path.home() / "Credenciais" / "grafana_credentials.json"
    OUTPUT_DIR = Path.home() / "Downloads"
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    CSV_PATH = OUTPUT_DIR / "resultado_dashboards.csv"
    LOG_PATH = OUTPUT_DIR / "verificacao_dashboards.log"

    # Configuração de log
    logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
    logging.StreamHandler(),
    logging.FileHandler(LOG_PATH, mode='w', encoding='utf-8')
    ]
    )

    def carregar_credenciais():
    try:
    with open(CREDENTIALS_PATH, "r", encoding="utf-8") as f:
    dados = json.load(f)
    return dados if isinstance(dados, list) else [dados]
    except Exception as e:
    logging.error(f"Erro ao carregar credenciais: {e}")
    return []

    def buscar_dashboards(cliente):
    url = f"{cliente['url']}/api/search?query=&type=dash-db"
    headers = {"Authorization": f"Bearer {cliente['api_key']}"}
    try:
    response = requests.get(url, headers=headers, timeout=15)
    response.raise_for_status()
    return response.json()
    except Exception as e:
    logging.error(f"Erro ao buscar dashboards de {cliente['nome']}: {e}")
    return []

    def buscar_dashboard_por_uid(cliente, uid):
    url = f"{cliente['url']}/api/dashboards/uid/{uid}"
    headers = {"Authorization": f"Bearer {cliente['api_key']}"}
    try:
    response = requests.get(url, headers=headers, timeout=15)
    response.raise_for_status()
    return response.json()
    except Exception as e:
    logging.error(f"Erro ao buscar dashboard UID {uid}: {e}")
    return {}

    def executar_query(cliente, datasource, target):
    url = f"{cliente['url']}/api/ds/query"
    headers = {
    "Authorization": f"Bearer {cliente['api_key']}",
    "Content-Type": "application/json"
    }
    payload = {
    "queries": [
    {
    "refId": "A",
    "datasource": datasource,
    "intervalMs": 60000,
    "maxDataPoints": 1,
    "target": target
    }
    ]
    }
    try:
    response = requests.post(url, headers=headers, json=payload, timeout=20)
    response.raise_for_status()
    data = response.json()
    result = data.get("results", {}).get("A", {})
    frames = result.get("frames", [])

    if not frames:
    return "No Data"
    return "OK"
    except Exception as e:
    logging.warning(f"Erro ao executar query: {e}")
    return "Erro"

    def extrair_paineis(paineis, row_name=None):
    todos = []
    for panel in paineis:
    if panel.get("type") == "row":
    nested = extrair_paineis(panel.get("panels", []), row_name=panel.get("title"))
    todos.extend(nested)
    else:
    panel["row_name"] = row_name
    todos.append(panel)
    return todos

    def diagnosticar():
    clientes = carregar_credenciais()
    if not clientes:
    return

    with open(CSV_PATH, "w", newline='', encoding="utf-8") as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(["Cliente", "Dashboard", "UID", "Aba", "Painel", "Retorno Data", "Url Painel"])

    for cliente in clientes:
    nome_cliente = cliente.get("nome", "")
    dashboards = buscar_dashboards(cliente)
    logging.info(f" Verificando cliente: {nome_cliente} - {len(dashboards)} dashboards encontrados")

    for dash in dashboards:
    uid = dash.get("uid")
    titulo_dash = dash.get("title")
    url_painel = f"{cliente['url']}/d/{uid}"

    dashboard_json = buscar_dashboard_por_uid(cliente, uid)
    folder_title = dashboard_json.get("meta", {}).get("folderTitle", "")
    dashboard_data = dashboard_json.get("dashboard", {})

    paineis = extrair_paineis(dashboard_data.get("panels", []))

    for panel in paineis:
    panel_title = panel.get("title", "Sem título")
    row_name = panel.get("row_name", "")
    targets = panel.get("targets", [])

    retorno = "N/A"
    if targets:
    for target in targets:
    ds = target.get("datasource")
    if ds:
    retorno = executar_query(cliente, ds, target)
    break

    logging.info(f"✔️ {folder_title} | {titulo_dash} | {uid} | {row_name} | {panel_title} | {retorno}")
    writer.writerow([
    folder_title, titulo_dash, uid, row_name, panel_title, retorno, url_painel
    ])

    if __name__ == "__main__":
    diagnosticar()```

    In the credentials file it looks like this:

    ```[
    {
    "nome": "Name",
    "url": "https://grafana.seudominio.com",
    "api_key": "seu_token_de_api"
    }
    ]

    Continue reading...

Compartilhe esta Página