Data Models and Types

@dataclass
class DatasourceIdentifier:
    """
    Identifies a Grafana data source by ID, UID, or name.
    
    Args:
        id (Optional[str]): Numerical data source ID (deprecated)
        uid (Optional[str]): Alphanumerical data source UID (recommended)
        name (Optional[str]): Data source name
    """
    id: Optional[str] = None
    uid: Optional[str] = None
    name: Optional[str] = None

@dataclass
class DatasourceModel:
    """
    Model for creating data source JSON payloads.
    
    Args:
        name (str): Data source name (must be unique)
        type (str): Data source type (prometheus, influxdb, mysql, etc.)
        url (str): Data source connection URL
        access (str): Access mode ("proxy" or "direct")
        database (Optional[str]): Database name (deprecated, use jsonData)
        user (Optional[str]): Username for authentication
        jsonData (Optional[Dict]): JSON configuration object
        secureJsonData (Optional[Dict]): Secure JSON data (passwords, tokens)
        secureJsonFields (Optional[Dict]): Secure JSON field configuration
    """
    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:
        """
        Convert to dictionary for API requests.
        
        Returns:
            Dict: Data source configuration as dictionary
        """
        ...

@dataclass
class DatasourceHealthResponse:
    """
    Response from data source health check operations.
    
    Args:
        uid (str): Data source UID
        type (Union[str, None]): Data source type
        success (bool): Health check success status
        status (str): Status message ("OK" or "ERROR")
        message (str): Detailed status message
        duration (Optional[float]): Request duration in seconds
        response (Optional[Any]): Full response object from health check
    """
    uid: str
    type: Union[str, None]
    success: bool
    status: str
    message: str
    duration: Optional[float] = None
    response: Optional[Any] = None
    
    def asdict(self) -> Dict:
        """
        Convert to dictionary including full response.
        
        Returns:
            Dict: Complete health response as dictionary
        """
        ...
    
    def asdict_compact(self) -> Dict:
        """
        Convert to dictionary without response object.
        
        Returns:
            Dict: Compact health response without raw response data
        """
        ...

from grafana_client import GrafanaApi, TokenAuth
from grafana_client.model import DatasourceModel, DatasourceIdentifier

api = GrafanaApi(auth=TokenAuth("your-token"), host="grafana.example.com")

# Create data source using model
prometheus_ds = DatasourceModel(
    name="Prometheus Production",
    type="prometheus",
    url="http://prometheus.prod.example.com:9090",
    access="proxy",
    jsonData={
        "httpMethod": "POST",
        "timeInterval": "5s",
        "queryTimeout": "60s",
        "disableMetricsLookup": False
    },
    secureJsonData={
        "httpHeaderValue1": "Bearer prod-token-12345"
    }
)

# Create data source
result = api.datasource.create_datasource(prometheus_ds.asdict())
created_uid = result['uid']
print(f"Created data source: {created_uid}")

# Use DatasourceIdentifier to retrieve data source
ds_identifier = DatasourceIdentifier(uid=created_uid)
datasource = api.datasource.get(ds_identifier)
print(f"Retrieved data source: {datasource['name']}")

# Alternative identification methods
ds_by_name = DatasourceIdentifier(name="Prometheus Production")
ds_by_id = DatasourceIdentifier(id="123")  # Deprecated

# Health check with response model
health_response = api.datasource.health_check(datasource)
print(f"Health check - Success: {health_response.success}")
print(f"Status: {health_response.status}")
print(f"Message: {health_response.message}")
print(f"Duration: {health_response.duration}s")

# Convert to dictionary for logging/serialization
health_data = health_response.asdict_compact()
print(f"Compact health data: {health_data}")

@dataclass
class PersonalPreferences:
    """
    User/team/organization preferences model.
    
    Args:
        homeDashboardId (Optional[int]): Home dashboard ID (deprecated)
        homeDashboardUID (Optional[str]): Home dashboard UID (recommended)
        locale (Optional[str]): Locale setting (e.g., "en-US", "de-DE")
        theme (Optional[str]): UI theme ("light", "dark", "auto")
        timezone (Optional[str]): Timezone (e.g., "browser", "utc", "America/New_York")
        weekStart (Optional[str]): Week start day ("monday", "sunday", "saturday")
    """
    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:
        """
        Convert to dictionary for API requests.
        
        Args:
            filter_none (bool): If True, exclude None values from output
            
        Returns:
            Dict: Preferences as dictionary
        """
        ...

from grafana_client.model import PersonalPreferences

# Create comprehensive user preferences
user_prefs = PersonalPreferences(
    homeDashboardUID="home-dashboard-uid-123",
    locale="en-US",
    theme="dark",
    timezone="America/New_York", 
    weekStart="monday"
)

# Update user preferences
api.user.update_preferences(user_prefs)
print("User preferences updated")

# Create partial preferences for team
team_prefs = PersonalPreferences(
    theme="light",
    timezone="UTC"
)

# Apply to team (only non-None values)
team_dict = team_prefs.asdict(filter_none=True)
api.teams.update_preferences(team_id=5, preferences=team_prefs)
print("Team preferences updated")

# Organization-wide default preferences
org_prefs = PersonalPreferences(
    theme="auto",
    timezone="browser",
    weekStart="monday",
    locale="en-US"
)

api.organization.update_preferences(org_prefs)
print("Organization default preferences set")

# Get and modify existing preferences
current_prefs = api.user.get_preferences()
print(f"Current theme: {current_prefs.theme}")
print(f"Current timezone: {current_prefs.timezone}")

# Partial update (patch)
theme_update = PersonalPreferences(theme="light")
api.user.patch_preferences(theme_update)
print("Theme updated to light")

class GrafanaException(Exception):
    """
    Base exception for all Grafana client errors.
    
    Args:
        status_code (int): HTTP status code from response
        response: Raw HTTP response object
        message (str): Human-readable error message
    """
    def __init__(self, status_code: int, response, message: str):
        self.status_code = status_code
        self.response = response
        self.message = message
        super().__init__(message)

class GrafanaTimeoutError(GrafanaException):
    """
    Raised when a request timeout occurs.
    Inherits from GrafanaException.
    """
    pass

class GrafanaServerError(GrafanaException):
    """
    Raised for 5xx HTTP server errors.
    Inherits from GrafanaException.
    """
    pass

class GrafanaClientError(GrafanaException):
    """
    Raised for 4xx HTTP client errors.
    Base class for client-side errors.
    """
    pass

class GrafanaBadInputError(GrafanaClientError):
    """
    Raised for 400 Bad Request errors.
    Indicates invalid input data or parameters.
    """
    pass

class GrafanaUnauthorizedError(GrafanaClientError):
    """
    Raised for 401 Unauthorized errors.
    Indicates authentication failure or invalid credentials.
    """
    pass

from grafana_client import (
    GrafanaApi, TokenAuth,
    GrafanaException, GrafanaTimeoutError, GrafanaServerError,
    GrafanaClientError, GrafanaBadInputError, GrafanaUnauthorizedError
)

api = GrafanaApi(
    auth=TokenAuth("your-token"),
    host="grafana.example.com",
    timeout=5.0
)

def robust_grafana_operation():
    """Example of comprehensive error handling"""
    try:
        # Attempt API operation
        dashboard = api.dashboard.get_dashboard("some-dashboard-uid")
        return dashboard
        
    except GrafanaTimeoutError as e:
        print(f"Operation timed out after {api.timeout}s: {e.message}")
        print(f"Status code: {e.status_code}")
        # Implement retry logic
        return None
        
    except GrafanaUnauthorizedError as e:
        print(f"Authentication failed: {e.message}")
        print("Check your API token and permissions")
        # Refresh token or prompt for new credentials
        return None
        
    except GrafanaBadInputError as e:
        print(f"Invalid request data: {e.message}")
        print(f"Response details: {e.response}")
        # Fix input data and retry
        return None
        
    except GrafanaClientError as e:
        if e.status_code == 404:
            print("Dashboard not found")
        elif e.status_code == 403:
            print("Access denied - insufficient permissions")
        else:
            print(f"Client error ({e.status_code}): {e.message}")
        return None
        
    except GrafanaServerError as e:
        print(f"Grafana server error ({e.status_code}): {e.message}")
        print("This may be a temporary server issue")
        # Implement exponential backoff retry
        return None
        
    except GrafanaException as e:
        print(f"General Grafana error ({e.status_code}): {e.message}")
        # Log error details and handle gracefully
        return None
        
    except Exception as e:
        print(f"Unexpected error: {e}")
        # Handle non-Grafana exceptions
        return None

# Use robust error handling
result = robust_grafana_operation()
if result:
    print("Operation succeeded")
else:
    print("Operation failed - see error messages above")

def setup_logging(level=None):
    """
    Setup logging configuration for the client.
    
    Args:
        level: Logging level (default: logging.INFO)
    """
    ...

def as_bool(value: str) -> bool:
    """
    Convert string representation to boolean.
    
    Args:
        value (str): String value to convert ("true", "false", "1", "0", etc.)
        
    Returns:
        bool: Converted boolean value
    """
    ...

def format_param_value(maybe_list):
    """
    Format parameter values for HTTP requests.
    
    Args:
        maybe_list: Value that may be a list or single value
        
    Returns:
        Formatted parameter value (comma-separated if list)
    """
    ...

# Constants
DEFAULT_TIMEOUT: float = 5.0
DEFAULT_SESSION_POOL_SIZE: int = 10

from grafana_client.util import setup_logging, as_bool, format_param_value
from grafana_client.client import DEFAULT_TIMEOUT, DEFAULT_SESSION_POOL_SIZE
import logging

# Setup logging for debugging
setup_logging(level=logging.DEBUG)
print("Debug logging enabled for grafana-client")

# Boolean conversion utility
config_values = {
    "enable_feature": "true",
    "debug_mode": "1", 
    "production": "false",
    "testing": "0"
}

for key, value in config_values.items():
    bool_value = as_bool(value)
    print(f"{key}: '{value}' -> {bool_value}")

# Parameter formatting for API requests
single_tag = "production"
multiple_tags = ["production", "monitoring", "alerts"]

formatted_single = format_param_value(single_tag)
formatted_multiple = format_param_value(multiple_tags)

print(f"Single tag: {formatted_single}")
print(f"Multiple tags: {formatted_multiple}")  # "production,monitoring,alerts"

# Use constants for configuration
print(f"Default timeout: {DEFAULT_TIMEOUT}s")
print(f"Default session pool size: {DEFAULT_SESSION_POOL_SIZE}")

# Custom client configuration using constants
custom_api = GrafanaApi(
    auth=TokenAuth("your-token"),
    timeout=DEFAULT_TIMEOUT * 2,  # Double the default timeout
    session_pool_size=DEFAULT_SESSION_POOL_SIZE * 2  # Larger pool
)

# Complete dashboard model example
dashboard_model = {
    "dashboard": {
        "id": None,
        "uid": "custom-dashboard-uid",
        "title": "Production Monitoring",
        "description": "Main production monitoring dashboard",
        "tags": ["production", "monitoring", "sre"],
        "timezone": "UTC",
        "editable": True,
        "hideControls": False,
        "schemaVersion": 30,
        "version": 1,
        "refresh": "30s",
        "time": {
            "from": "now-24h",
            "to": "now"
        },
        "timepicker": {
            "refresh_intervals": ["10s", "30s", "1m", "5m", "15m", "30m", "1h"],
            "time_options": ["5m", "15m", "1h", "6h", "12h", "24h", "2d", "7d", "30d"]
        },
        "templating": {
            "list": [
                {
                    "name": "instance",
                    "type": "query",
                    "datasource": {"uid": "prometheus-uid"},
                    "query": "label_values(up, instance)",
                    "multi": True,
                    "includeAll": True,
                    "allValue": ".*"
                }
            ]
        },
        "panels": [
            {
                "id": 1,
                "title": "CPU Usage",
                "type": "stat",
                "gridPos": {"h": 8, "w": 12, "x": 0, "y": 0},
                "fieldConfig": {
                    "defaults": {
                        "unit": "percent",
                        "min": 0,
                        "max": 100,
                        "thresholds": {
                            "steps": [
                                {"color": "green", "value": None},
                                {"color": "yellow", "value": 70},
                                {"color": "red", "value": 90}
                            ]
                        }
                    }
                },
                "targets": [
                    {
                        "expr": "100 - (avg(irate(node_cpu_seconds_total{mode=\"idle\",instance=~\"$instance\"}[5m])) * 100)",
                        "refId": "A",
                        "datasource": {"uid": "prometheus-uid"}
                    }
                ]
            }
        ]
    },
    "folderId": 0,
    "folderUID": "general",
    "message": "Updated via API",
    "overwrite": True
}

# Modern alert rule model example
alert_rule_model = {
    "uid": "",  # Auto-generated
    "title": "High CPU Alert",
    "condition": "B",
    "data": [
        {
            "refId": "A",
            "queryType": "",
            "relativeTimeRange": {"from": 300, "to": 0},
            "datasourceUid": "prometheus-uid",
            "model": {
                "expr": "avg(cpu_usage_percent) by (instance)",
                "interval": "",
                "refId": "A"
            }
        },
        {
            "refId": "B",
            "queryType": "",
            "relativeTimeRange": {"from": 0, "to": 0},
            "datasourceUid": "__expr__",
            "model": {
                "expression": "A",
                "reducer": "last",
                "type": "reduce",
                "refId": "B"
            }
        }
    ],
    "folderUID": "alerts",
    "ruleGroup": "Infrastructure",
    "noDataState": "NoData",
    "execErrState": "Alerting",
    "for": "5m",
    "annotations": {
        "description": "CPU usage is above threshold",
        "runbook_url": "https://wiki.example.com/cpu-alerts",
        "summary": "High CPU on {{ $labels.instance }}"
    },
    "labels": {
        "severity": "warning",
        "team": "sre",
        "component": "infrastructure"
    }
}