CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-grafana-client

A client library for accessing the Grafana HTTP API, written in Python

—

Pending
Overview
Eval results
Files

client-management.mddocs/

Client Management

Core client classes for both synchronous and asynchronous operations, providing connection management, error handling, configuration options, and factory methods for convenient instantiation of Grafana API clients.

Capabilities

Synchronous API Client

Main synchronous client class providing access to all Grafana API elements through a unified interface.

class GrafanaApi:
    """
    Main entry point for synchronous Grafana API operations.
    
    Args:
        auth: Authentication credentials (TokenAuth, HeaderAuth, tuple, or AuthBase)
        host (str): Grafana host (default: "localhost")
        port (Optional[int]): Grafana port (optional, inferred from protocol)
        url_path_prefix (str): URL path prefix (default: "")
        protocol (str): Protocol (default: "http")
        verify (bool): SSL verification (default: True)
        timeout (float): Request timeout in seconds (default: 5.0)
        user_agent (Optional[str]): Custom user agent string (optional)
        organization_id (Optional[int]): Organization ID to use (optional)
        session_pool_size (int): HTTP session pool size (default: 10)
    """
    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:
        """Connect to Grafana and check health"""
        ...
    
    @property
    def version(self) -> str:
        """Get Grafana version"""
        ...
    
    @classmethod
    def from_url(cls, url: str = None, 
                 credential: Union[str, Tuple[str, str], niquests.auth.AuthBase] = None, 
                 timeout: Union[float, Tuple[float, float]] = 5.0) -> 'GrafanaApi':
        """
        Create GrafanaApi instance from URL.
        
        Args:
            url (str): Grafana URL with optional embedded credentials (default: "http://admin:admin@localhost:3000")
            credential: Optional credential (token string, (username, password) tuple, or AuthBase instance)
            timeout (Union[float, Tuple[float, float]]): Request timeout or (connect, read) timeouts (default: 5.0)
        
        Returns:
            GrafanaApi: Configured API instance
        """
        ...
    
    @classmethod  
    def from_env(cls, timeout: Union[float, Tuple[float, float]] = None) -> 'GrafanaApi':
        """
        Create GrafanaApi instance from environment variables.
        
        Args:
            timeout (Union[float, Tuple[float, float]]): Request timeout or (connect, read) timeouts (default: determined by GRAFANA_TIMEOUT or 5.0)
            
        Returns:
            GrafanaApi: Configured API instance from environment
        """
        ...
    
    # Element API attributes (all synchronous versions)
    @property
    def admin(self) -> Admin: ...
    
    @property
    def alerting(self) -> Alerting: ...
    
    @property
    def alertingprovisioning(self) -> AlertingProvisioning: ...
    
    @property
    def annotations(self) -> Annotations: ...
    
    @property
    def dashboard(self) -> Dashboard: ...
    
    @property
    def dashboard_versions(self) -> DashboardVersions: ...
    
    @property
    def datasource(self) -> Datasource: ...
    
    @property
    def folder(self) -> Folder: ...
    
    @property
    def health(self) -> Health: ...
    
    @property
    def libraryelement(self) -> LibraryElement: ...
    
    @property
    def notifications(self) -> Notifications: ...
    
    @property
    def organization(self) -> Organization: ...
    
    @property
    def organizations(self) -> Organizations: ...
    
    @property
    def plugin(self) -> Plugin: ...
    
    @property
    def rbac(self) -> Rbac: ...
    
    @property
    def search(self) -> Search: ...
    
    @property
    def serviceaccount(self) -> ServiceAccount: ...
    
    @property
    def snapshots(self) -> Snapshots: ...
    
    @property
    def teams(self) -> Teams: ...
    
    @property
    def user(self) -> User: ...
    
    @property
    def users(self) -> Users: ...

Basic Usage:

from grafana_client import GrafanaApi, TokenAuth

# Create client with token authentication
auth = TokenAuth(token="your-api-token")
api = GrafanaApi(
    auth=auth,
    host="grafana.example.com",
    protocol="https", 
    timeout=10.0,
    organization_id=1
)

# Connect and verify
api.connect()
print(f"Connected to Grafana {api.version}")

# Access API elements
dashboards = api.dashboard.get_dashboards_tags()
datasources = api.datasource.list_datasources()
users = api.users.search_users()

Asynchronous API Client

Main asynchronous client class providing the same API elements as the synchronous version but with async/await support.

class AsyncGrafanaApi:
    """
    Main entry point for asynchronous Grafana API operations.
    
    Args:
        auth: Authentication credentials (TokenAuth, HeaderAuth, tuple, or AuthBase)
        host (str): Grafana host (default: "localhost")  
        port (Optional[int]): Grafana port (optional, inferred from protocol)
        url_path_prefix (str): URL path prefix (default: "")
        protocol (str): Protocol (default: "http")
        verify (bool): SSL verification (default: True)
        timeout (float): Request timeout in seconds (default: 5.0)
        user_agent (Optional[str]): Custom user agent string (optional)
        organization_id (Optional[int]): Organization ID to use (optional)
    """
    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:
        """Connect to Grafana and check health"""
        ...
    
    @property
    async def version(self) -> str:
        """Get Grafana version"""
        ...
    
    @classmethod
    def from_url(cls, url: str, credential = None, timeout: float = 5.0) -> 'AsyncGrafanaApi':
        """Create AsyncGrafanaApi instance from URL"""
        ...
    
    @classmethod
    def from_env(cls, timeout: float = 5.0) -> 'AsyncGrafanaApi':
        """Create AsyncGrafanaApi instance from environment variables"""
        ...
    
    # Element API attributes (all asynchronous versions)
    @property
    def admin(self) -> AsyncAdmin: ...
    
    @property
    def alerting(self) -> AsyncAlerting: ...
    
    @property
    def alertingprovisioning(self) -> AsyncAlertingProvisioning: ...
    
    @property
    def annotations(self) -> AsyncAnnotations: ...
    
    @property
    def dashboard(self) -> AsyncDashboard: ...
    
    @property
    def dashboard_versions(self) -> AsyncDashboardVersions: ...
    
    @property
    def datasource(self) -> AsyncDatasource: ...
    
    @property
    def folder(self) -> AsyncFolder: ...
    
    @property
    def health(self) -> AsyncHealth: ...
    
    @property
    def libraryelement(self) -> AsyncLibraryElement: ...
    
    @property
    def notifications(self) -> AsyncNotifications: ...
    
    @property
    def organization(self) -> AsyncOrganization: ...
    
    @property
    def organizations(self) -> AsyncOrganizations: ...
    
    @property
    def plugin(self) -> AsyncPlugin: ...
    
    @property
    def rbac(self) -> AsyncRbac: ...
    
    @property
    def search(self) -> AsyncSearch: ...
    
    @property
    def serviceaccount(self) -> AsyncServiceAccount: ...
    
    @property
    def snapshots(self) -> AsyncSnapshots: ...
    
    @property
    def teams(self) -> AsyncTeams: ...
    
    @property
    def user(self) -> AsyncUser: ...
    
    @property
    def users(self) -> AsyncUsers: ...

Basic Async Usage:

import asyncio
from grafana_client import AsyncGrafanaApi, TokenAuth

async def main():
    # Create async client
    auth = TokenAuth(token="your-api-token")
    api = AsyncGrafanaApi(
        auth=auth,
        host="grafana.example.com",
        protocol="https",
        timeout=10.0
    )
    
    # Connect and verify
    await api.connect()
    version = await api.version
    print(f"Connected to Grafana {version}")
    
    # Access async API elements
    dashboards = await api.dashboard.get_dashboards_tags()
    datasources = await api.datasource.list_datasources()
    users = await api.users.search_users()

# Run async code
asyncio.run(main())

Low-Level HTTP Clients

Direct HTTP client classes for advanced use cases requiring fine-grained control over requests.

class GrafanaClient:
    """
    Low-level synchronous HTTP client for Grafana API.
    
    Args:
        auth: Authentication credentials
        host (str): Grafana host (default: "localhost")
        port (Optional[int]): Grafana port (optional)
        url_path_prefix (str): URL path prefix (default: "")
        protocol (str): Protocol (default: "http")
        verify (bool): SSL verification (default: True)
        timeout (float): Request timeout
        user_agent (Optional[str]): Custom user agent string
        organization_id (Optional[int]): Organization ID to use
        session_pool_size (int): HTTP session pool size
    """
    def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
                 url_path_prefix: str = "", protocol: str = "http", 
                 verify: bool = True, timeout: float = None,
                 user_agent: Optional[str] = None, organization_id: Optional[int] = None,
                 session_pool_size: int = 10): ...

class AsyncGrafanaClient:
    """
    Low-level asynchronous HTTP client for Grafana API.
    
    Same constructor parameters as GrafanaClient.
    """
    def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
                 url_path_prefix: str = "", protocol: str = "http",
                 verify: bool = True, timeout: float = None,
                 user_agent: Optional[str] = None, organization_id: Optional[int] = None): ...

Factory Methods

Convenient methods for creating client instances from URLs or environment variables.

URL-based Factory:

# URL with embedded credentials
api = GrafanaApi.from_url("https://admin:password@grafana.example.com:3000/grafana")

# URL with separate credential
api = GrafanaApi.from_url(
    url="https://grafana.example.com:3000",
    credential="your-api-token",
    timeout=15.0
)

# Custom path prefix
api = GrafanaApi.from_url("https://grafana.example.com/custom-path")

Environment-based Factory:

Set environment variables:

export GRAFANA_HOST=grafana.example.com
export GRAFANA_PROTOCOL=https
export GRAFANA_TOKEN=your-api-token
export GRAFANA_ORGANIZATION_ID=1
export GRAFANA_TIMEOUT=10.0
export GRAFANA_VERIFY_SSL=true

Create client:

# Automatically configured from environment
api = GrafanaApi.from_env()

# With custom timeout
api = GrafanaApi.from_env(timeout=20.0)

Connection Management

Health Checking:

from grafana_client import GrafanaApi, GrafanaException

try:
    api = GrafanaApi(auth=auth, host="grafana.example.com")
    api.connect()  # Validates connection and health
    print("Grafana is healthy")
except GrafanaException as e:
    print(f"Connection failed: {e.message}")

Timeout Configuration:

# Global timeout for all requests
api = GrafanaApi(auth=auth, timeout=30.0)

# Different timeouts for different operations
api.connect()  # Uses global timeout

# Individual request timeouts handled by underlying HTTP client

SSL Configuration:

# Disable SSL verification (not recommended for production)
api = GrafanaApi(auth=auth, protocol="https", verify=False)

# Custom SSL configuration through requests
import requests
session = requests.Session()
session.verify = "/path/to/ca-bundle.crt"
# Use custom session through low-level client if needed

Organization Context

Setting Organization Context:

# Set organization during client creation
api = GrafanaApi(auth=auth, organization_id=5)

# Switch organization context at runtime
api.organizations.switch_organization(organization_id=3)

# Get current organization
current_org = api.organization.get_current_organization()
print(f"Current organization: {current_org['name']}")

Session Management

HTTP Session Pool Configuration:

# Configure connection pooling for high-throughput scenarios
api = GrafanaApi(
    auth=auth,
    session_pool_size=50,  # Larger pool for concurrent requests
    timeout=5.0
)

# Async API doesn't use session pool (uses aiohttp)
async_api = AsyncGrafanaApi(auth=auth, timeout=5.0)

Error Handling

Connection and client-level errors:

from grafana_client import (
    GrafanaApi, GrafanaException, GrafanaTimeoutError, 
    GrafanaServerError, GrafanaClientError
)

try:
    api = GrafanaApi(auth=auth, host="invalid-host", timeout=2.0)
    api.connect()
except GrafanaTimeoutError as e:
    print(f"Connection timeout: {e}")
except GrafanaServerError as e:
    print(f"Server error ({e.status_code}): {e.message}")
except GrafanaClientError as e:
    print(f"Client error ({e.status_code}): {e.message}")
except GrafanaException as e:
    print(f"General error: {e}")

Performance Considerations

Synchronous vs Asynchronous:

# Use sync API for simple scripts and single-threaded applications
api = GrafanaApi(auth=auth)
result = api.dashboard.get_dashboard("dashboard-uid")

# Use async API for concurrent operations and high-throughput scenarios  
async def fetch_multiple_dashboards():
    api = AsyncGrafanaApi(auth=auth)
    tasks = [
        api.dashboard.get_dashboard(uid) 
        for uid in dashboard_uids
    ]
    results = await asyncio.gather(*tasks)
    return results

Connection Reuse:

# Reuse the same client instance for multiple operations
api = GrafanaApi(auth=auth)
api.connect()  # Establish connection once

# Multiple operations reuse the connection
dashboard1 = api.dashboard.get_dashboard("uid1")
dashboard2 = api.dashboard.get_dashboard("uid2")  
datasources = api.datasource.list_datasources()

Constants

DEFAULT_TIMEOUT: float = 5.0
DEFAULT_SESSION_POOL_SIZE: int = 10

Install with Tessl CLI

npx tessl i tessl/pypi-grafana-client@5.0.1

docs

admin-and-rbac.md

alerting.md

authentication.md

client-management.md

dashboards.md

data-models.md

datasources.md

index.md

library-elements.md

plugin-management.md

snapshot-management.md

users-and-orgs.md

tile.json