python获取grafana

admin 2026年02月20日 03:19 102 0

通过Python可自动化获取Grafana监控数据与配置信息，主要依赖其REST API，使用requests库或grafana-api第三方库，通过API Key认证后，可请求仪表盘列表（/api/dashboards）、面板数据（/api/dashboards/db/{uid}）、数据源信息（/api/datasources）等接口，解析返回的JSON数据，提取面板指标、图表配置、查询语句等关键信息，适用于监控系统集成、自动化报表生成或环境巡检场景，提升运维效率与数据处理的灵活性。

使用Python高效获取Grafana数据：API调用实战指南

Grafana作为业界领先的开源监控可视化平台，在系统监控、业务数据分析等场景中扮演着核心角色，在实际工程实践中，我们常需通过编程方式自动化获取Grafana中的仪表盘、数据源、监控指标等关键数据，以实现定期数据导出、智能报表生成、监控告警自动化等目标，Python凭借其简洁的语法、强大的第三方库生态及灵活的扩展能力，成为与Grafana API交互的理想选择，本文将系统介绍通过Python调用Grafana API的完整方案，涵盖核心概念、实战代码及典型应用场景。

Grafana API核心机制

在通过Python与Grafana API交互前，需深入理解其核心设计原则，Grafana提供RESTful风格的HTTP API，支持通过标准HTTP方法进行资源操作,主要包含以下关键要素：

认证机制详解

Grafana API提供两种主流认证方式,需根据场景安全选择：

API Key认证（推荐）：适用于服务端自动化脚本，具有较高安全性，在Grafana界面（Configuration → API Keys）创建Key时，建议按需分配"Editor"或"Admin"权限,避免过度授权。
Bearer Token认证：适用于临时性访问场景，如用户登录后的会话Token,需注意Token的生命周期管理。

核心API端点解析

Grafana API采用模块化设计,常用端点分类如下：

仪表盘管理： /api/dashboards（获取/创建仪表盘列表） /api/dashboards/{uid}（获取特定仪表盘完整配置）
数据源管理： /api/datasources（数据源CRUD操作）
用户管理： /api/users（用户列表管理）
数据查询： /api/datasources/{id}/query（执行数据查询,需适配具体数据源类型如Prometheus）

API版本演进

Grafana 7+版本推荐使用/api前缀的v1/v2稳定API，旧版API（如/api/db）已逐步废弃,本文所有示例均基于最新稳定版API实现。

Python实现：Grafana数据获取全流程

环境准备

首先安装核心HTTP请求库：

pip install requests

API客户端封装

设计具备会话管理、错误重试功能的GrafanaClient类：

import requests
import json
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class GrafanaClient:
def init(self, grafana_url, api_key=None, timeout=30):
self.grafana_url = grafana_url.rstrip('/')
self.api_key = api_key
self.timeout = timeout
self.session = requests.Session()
    # 配置重试策略
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    self.session.mount("http://", adapter)
    self.session.mount("https://", adapter)
    # 设置请求头
    self.headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}' if api_key else None
    }
def _make_request(self, method, endpoint, params=None, data=None):
    """发送HTTP请求并处理异常"""
    url = f"{self.grafana_url}{endpoint}"
    try:
        response = self.session.request(
            method=method,
            url=url,
            headers=self.headers,
            params=params,
            data=json.dumps(data) if data else None,
            timeout=self.timeout
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        raise Exception(f"Grafana API请求失败 [{method} {url}]: {str(e)}")
def get_dashboards(self):
    """获取所有仪表盘元数据"""
    return self._make_request('GET', '/api/dashboards')
def get_dashboard_by_uid(self, uid):
    """通过UID获取仪表盘完整配置"""
    return self._make_request('GET', f'/api/dashboards/{uid}')
def get_datasources(self):
    """获取所有数据源配置"""
    return self._make_request('GET', '/api/datasources')
def query_data(self, datasource_id, query, time_range=None):
    """执行数据查询（需适配具体数据源类型）"""
    payload = {
        "queries": query if isinstance(query, list) else [query]
    }
    if time_range:
        payload["timeRange"] = time_range
    return self._make_request('POST', f'/api/datasources/{datasource_id}/query', data=payload)

实战示例：仪表盘数据获取

# 初始化客户端
client = GrafanaClient(
    grafana_url="http://localhost:3000",
    api_key="glsa_YourApiKeyHere"  # 建议从环境变量读取
)
获取仪表盘列表并解析
try:
dashboards = client.get_dashboards()
print("=== 仪表盘列表 ===")
for db in dashboards.get('dashboards', []):
print(f"- 标题: {db['title']}\n  UID: {db['uid']}\n  版本: {db['version']}")
# 获取特定仪表盘详情
target_uid = "example-dashboard-uid"
dashboard_detail = client.get_dashboard_by_uid(target_uid)
print(f"\n=== 仪表盘详情 [UID: {target_uid}] ===")
panels = dashboard_detail['dashboard']['panels']
print(f"标题: {dashboard_detail['dashboard']['title']}")
print(f"面板数量: {len(panels)}")
print(f"更新时间: {dashboard_detail['dashboard']['updated']}")
except Exception as e:
print(f"操作失败: {str(e)}")

实战示例：数据源管理

标签： #python grafana #获 #取API