A client library for accessing the Grafana HTTP API, written in Python
npx @tessl/cli install tessl/pypi-grafana-client@5.0.0A comprehensive Python client library for accessing the Grafana HTTP API. The grafana-client package provides both synchronous and asynchronous interfaces for managing dashboards, data sources, users, organizations, alerting, and all other aspects of Grafana via its REST API.
pip install grafana-clientfrom grafana_client import GrafanaApi, AsyncGrafanaApi
from grafana_client import TokenAuth, HeaderAuthCommon for working with the API:
from grafana_client import GrafanaApi
from grafana_client.model import DatasourceModel, DatasourceIdentifier, PersonalPreferences, DatasourceHealthResponsefrom grafana_client import GrafanaApi, TokenAuth
# Create API client with token authentication
auth = TokenAuth(token="your-grafana-api-token")
api = GrafanaApi(auth=auth, host="your-grafana-host")
# Connect and check health
api.connect()
print(f"Grafana version: {api.version}")
# Create a dashboard
dashboard_data = {
"dashboard": {
"title": "My Dashboard",
"panels": [],
"tags": ["example"]
}
}
result = api.dashboard.update_dashboard(dashboard_data)
print(f"Dashboard created: {result}")
# List data sources
datasources = api.datasource.list_datasources()
for ds in datasources:
print(f"Data source: {ds['name']} ({ds['type']})")import asyncio
from grafana_client import AsyncGrafanaApi, TokenAuth
async def main():
# Create async API client
auth = TokenAuth(token="your-grafana-api-token")
api = AsyncGrafanaApi(auth=auth, host="your-grafana-host")
# Connect and check health
await api.connect()
version = await api.version
print(f"Grafana version: {version}")
# Create a dashboard
dashboard_data = {
"dashboard": {
"title": "My Async Dashboard",
"panels": [],
"tags": ["async", "example"]
}
}
result = await api.dashboard.update_dashboard(dashboard_data)
print(f"Dashboard created: {result}")
# List data sources
datasources = await api.datasource.list_datasources()
for ds in datasources:
print(f"Data source: {ds['name']} ({ds['type']})")
# Run async code
asyncio.run(main())The grafana-client follows an element-based architecture that mirrors Grafana's API structure:
Each major Grafana API area is represented by an element class accessible as an attribute:
api.dashboard - Dashboard operationsapi.datasource - Data source managementapi.user - Current user operationsapi.users - Multi-user operationsapi.teams - Team managementapi.organization - Current organizationapi.organizations - Multi-organization operationsapi.admin - Administrative functionsapi.alerting - Legacy alertingapi.alertingprovisioning - Modern alerting provisioningapi.rbac - Role-based access controlThis design provides a clean, organized API surface while maintaining full access to Grafana's extensive functionality.
Token-based and custom header authentication for secure API access, with support for basic authentication and custom authentication schemes.
class TokenAuth:
def __init__(self, token: str): ...
class HeaderAuth:
def __init__(self, name: str, value: str): ...Core client classes for both synchronous and asynchronous operations, connection management, error handling, and factory methods for convenient instantiation.
class GrafanaApi:
def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
url_path_prefix: str = "", protocol: str = "http",
verify: bool = True, timeout: float = 5.0,
user_agent: Optional[str] = None, organization_id: Optional[int] = None,
session_pool_size: int = 10): ...
def connect(self) -> None: ...
@property
def version(self) -> str: ...
@classmethod
def from_url(cls, url: str, credential, timeout: float = 5.0) -> 'GrafanaApi': ...
@classmethod
def from_env(cls, timeout: float = 5.0) -> 'GrafanaApi': ...
class AsyncGrafanaApi:
def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
url_path_prefix: str = "", protocol: str = "http",
verify: bool = True, timeout: float = 5.0,
user_agent: Optional[str] = None, organization_id: Optional[int] = None): ...
async def connect(self) -> None: ...
@property
async def version(self) -> str: ...Complete dashboard lifecycle management including creation, updates, deletion, permissions, versioning, and search capabilities.
def get_dashboard(self, dashboard_uid: str): ...
def update_dashboard(self, dashboard: dict): ...
def delete_dashboard(self, dashboard_uid: str): ...
def get_permissions_by_uid(self, dashboard_uid: str): ...
def update_permissions_by_uid(self, dashboard_uid: str, items: list): ...
def get_dashboards_tags(self): ...Comprehensive data source operations including CRUD operations, health checks, query execution, and smart query capabilities with type-safe data models.
def list_datasources(self): ...
def get_datasource_by_uid(self, datasource_uid: str): ...
def create_datasource(self, datasource: dict): ...
def update_datasource_by_uid(self, datasource_uid: str, datasource: dict): ...
def delete_datasource_by_uid(self, datasource_uid: str): ...
def health(self, datasource_uid: str): ...
def query(self, datasource_id: int, query: dict, timestamp: Optional[int] = None): ...
def health_check(self, datasource: dict): ...User lifecycle management, team operations, organization administration, and preference management with comprehensive search and filtering capabilities.
def get_actual_user(self): ...
def search_users(self, query: Optional[str] = None, page: int = 1, perpage: int = 1000): ...
def get_user(self, user_id: int): ...
def find_user(self, login_or_email: str): ...
def search_teams(self, query: Optional[str] = None, page: int = 1, perpage: int = 1000): ...
def get_current_organization(self): ...
def list_organization(self): ...Legacy alerting, modern alerting provisioning, notification channels, contact points, notification policies, and mute timings management.
def get_alertrule(self, folder_name: str, alertrule_name: str): ...
def create_alertrule(self, folder_name: str, alertrule: dict): ...
def get_alertrules_all(self): ...
def get_contactpoints(self, name: Optional[str] = None): ...
def create_contactpoint(self, contactpoint: dict, disable_provenance: bool = False): ...
def get_notification_policy_tree(self): ...Administrative functions, user management, role-based access control, service accounts, and system-level operations with comprehensive permission management.
def settings(self): ...
def stats(self): ...
def create_user(self, user: dict): ...
def change_user_permissions(self, user_id: int, is_grafana_admin: bool): ...
def get_rbac_roles_all(self): ...
def set_rbac_datasources_teams(self, datasource_uid: str, team_id: int, permission: str): ...Organization and management of dashboard folders with hierarchical structure support, including creation, updates, permissions, and nested folder operations.
def get_all_folders(self, parent_uid: Optional[str] = None): ...
def get_folder(self, uid: str): ...
def get_folder_by_id(self, folder_id: int): ...
def create_folder(self, title: str, uid: Optional[str] = None, parent_uid: Optional[str] = None): ...
def move_folder(self, uid: str, parent_uid: str): ...
def update_folder(self, uid: str, title: Optional[str] = None, version: Optional[int] = None,
overwrite: bool = False, new_uid: Optional[str] = None): ...
def delete_folder(self, uid: str): ...
def get_folder_permissions(self, uid: str): ...
def update_folder_permissions(self, uid: str, items: list): ...
def update_folder_permissions_for_user(self, uid: str, user_id: int, items: dict): ...Comprehensive search functionality for dashboards, folders, and other resources with advanced filtering, tagging, and pagination capabilities.
def search_dashboards(self, query: Optional[str] = None, tag: Optional[Union[str, list]] = None,
type_: Optional[str] = None, dashboard_ids: Optional[list] = None,
dashboard_uids: Optional[list] = None, folder_ids: Optional[list] = None,
folder_uids: Optional[list] = None, starred: Optional[bool] = None,
limit: Optional[int] = None, page: Optional[int] = None): ...Service account lifecycle management for programmatic access, including creation, token management, permissions, and integration with authentication systems.
def get(self, service_account_id: int): ...
def create(self, service_account: dict): ...
def update(self, service_account_id: int, service_account: dict): ...
def delete(self, service_account_id: int): ...
def search(self, query: Optional[str] = None, page: Optional[int] = None, perpage: Optional[int] = None): ...
def search_one(self, service_account_name: str): ...
def get_tokens(self, service_account_id: int): ...
def create_token(self, service_account_id: int, content: dict): ...
def delete_token(self, service_account_id: int, service_account_token_id: int): ...Plugin lifecycle management including listing, installation, uninstallation, health checks, and metrics collection for Grafana plugins.
def list(self): ...
def by_id(self, plugin_id: str): ...
def install(self, plugin_id: str, version: Optional[str] = None, errors: str = "raise"): ...
def uninstall(self, plugin_id: str, errors: str = "raise"): ...
def health(self, plugin_id: str): ...
def metrics(self, plugin_id: str): ...Dashboard snapshot creation and management for sharing and archiving dashboard states with configurable expiration and external storage options.
def create_new_snapshot(self, dashboard: dict, name: Optional[str] = None,
expires: Optional[int] = None, external: Optional[bool] = None,
key: Optional[str] = None, delete_key: Optional[str] = None): ...
def get_dashboard_snapshots(self): ...
def get_snapshot_by_key(self, key: str): ...
def delete_snapshot_by_key(self, key: str): ...
def delete_snapshot_by_delete_key(self, delete_key: str): ...Management of reusable library panels and variables that can be shared across multiple dashboards. Available in Grafana 8.2+.
def get_library_element(self, element_uid: str): ...
def get_library_element_by_name(self, element_name: str): ...
def get_library_element_connections(self, element_uid: str): ...
def create_library_element(self, model: dict, name: Optional[str] = None,
kind: int = 1, uid: Optional[str] = None,
folder_uid: Optional[str] = None): ...
def update_library_element(self, uid: str, model: dict, name: Optional[str] = None,
kind: int = 1, folder_uid: Optional[str] = None,
version: Optional[int] = None): ...
def delete_library_element(self, element_uid: str): ...
def list_library_elements(self, search_string: Optional[str] = None,
kind: Optional[int] = None, sort_direction: Optional[str] = None,
type_filter: Optional[str] = None, exclude_uid: Optional[str] = None,
folder_filter: Optional[str] = None, per_page: Optional[int] = None,
page: Optional[int] = None): ...Type-safe data classes for API payloads, responses, and configuration objects with validation and serialization support.
@dataclass
class DatasourceModel:
name: str
type: str
url: str
access: str
database: Optional[str] = None
user: Optional[str] = None
jsonData: Optional[Dict] = None
secureJsonData: Optional[Dict] = None
secureJsonFields: Optional[Dict] = None
def asdict(self) -> Dict: ...
@dataclass
class DatasourceIdentifier:
id: Optional[str] = None
uid: Optional[str] = None
name: Optional[str] = None
@dataclass
class PersonalPreferences:
homeDashboardId: Optional[int] = None
homeDashboardUID: Optional[str] = None
locale: Optional[str] = None
theme: Optional[str] = None
timezone: Optional[str] = None
weekStart: Optional[str] = None
def asdict(self, filter_none: bool = False) -> Dict: ...The library provides a comprehensive exception hierarchy for robust error handling:
class GrafanaException(Exception):
def __init__(self, status_code: int, response, message: str): ...
class GrafanaTimeoutError(GrafanaException): ...
class GrafanaServerError(GrafanaException): ...
class GrafanaClientError(GrafanaException): ...
class GrafanaBadInputError(GrafanaClientError): ...
class GrafanaUnauthorizedError(GrafanaClientError): ...All API operations can raise these exceptions based on HTTP response codes and error conditions.