0
# Data Models and Types
1
2
Type-safe data classes for API payloads, responses, and configuration objects with validation and serialization support. These models provide structured interfaces for complex API interactions and ensure type safety when working with Grafana API data.
3
4
## Capabilities
5
6
### Data Source Models
7
8
Type-safe models for data source configuration, identification, and health checking.
9
10
```python { .api }
11
@dataclass
12
class DatasourceIdentifier:
13
"""
14
Identifies a Grafana data source by ID, UID, or name.
15
16
Args:
17
id (Optional[str]): Numerical data source ID (deprecated)
18
uid (Optional[str]): Alphanumerical data source UID (recommended)
19
name (Optional[str]): Data source name
20
"""
21
id: Optional[str] = None
22
uid: Optional[str] = None
23
name: Optional[str] = None
24
25
@dataclass
26
class DatasourceModel:
27
"""
28
Model for creating data source JSON payloads.
29
30
Args:
31
name (str): Data source name (must be unique)
32
type (str): Data source type (prometheus, influxdb, mysql, etc.)
33
url (str): Data source connection URL
34
access (str): Access mode ("proxy" or "direct")
35
database (Optional[str]): Database name (deprecated, use jsonData)
36
user (Optional[str]): Username for authentication
37
jsonData (Optional[Dict]): JSON configuration object
38
secureJsonData (Optional[Dict]): Secure JSON data (passwords, tokens)
39
secureJsonFields (Optional[Dict]): Secure JSON field configuration
40
"""
41
name: str
42
type: str
43
url: str
44
access: str
45
database: Optional[str] = None
46
user: Optional[str] = None
47
jsonData: Optional[Dict] = None
48
secureJsonData: Optional[Dict] = None
49
secureJsonFields: Optional[Dict] = None
50
51
def asdict(self) -> Dict:
52
"""
53
Convert to dictionary for API requests.
54
55
Returns:
56
Dict: Data source configuration as dictionary
57
"""
58
...
59
60
@dataclass
61
class DatasourceHealthResponse:
62
"""
63
Response from data source health check operations.
64
65
Args:
66
uid (str): Data source UID
67
type (Union[str, None]): Data source type
68
success (bool): Health check success status
69
status (str): Status message ("OK" or "ERROR")
70
message (str): Detailed status message
71
duration (Optional[float]): Request duration in seconds
72
response (Optional[Any]): Full response object from health check
73
"""
74
uid: str
75
type: Union[str, None]
76
success: bool
77
status: str
78
message: str
79
duration: Optional[float] = None
80
response: Optional[Any] = None
81
82
def asdict(self) -> Dict:
83
"""
84
Convert to dictionary including full response.
85
86
Returns:
87
Dict: Complete health response as dictionary
88
"""
89
...
90
91
def asdict_compact(self) -> Dict:
92
"""
93
Convert to dictionary without response object.
94
95
Returns:
96
Dict: Compact health response without raw response data
97
"""
98
...
99
```
100
101
**Data Source Models Usage Example:**
102
103
```python
104
from grafana_client import GrafanaApi, TokenAuth
105
from grafana_client.model import DatasourceModel, DatasourceIdentifier
106
107
api = GrafanaApi(auth=TokenAuth("your-token"), host="grafana.example.com")
108
109
# Create data source using model
110
prometheus_ds = DatasourceModel(
111
name="Prometheus Production",
112
type="prometheus",
113
url="http://prometheus.prod.example.com:9090",
114
access="proxy",
115
jsonData={
116
"httpMethod": "POST",
117
"timeInterval": "5s",
118
"queryTimeout": "60s",
119
"disableMetricsLookup": False
120
},
121
secureJsonData={
122
"httpHeaderValue1": "Bearer prod-token-12345"
123
}
124
)
125
126
# Create data source
127
result = api.datasource.create_datasource(prometheus_ds.asdict())
128
created_uid = result['uid']
129
print(f"Created data source: {created_uid}")
130
131
# Use DatasourceIdentifier to retrieve data source
132
ds_identifier = DatasourceIdentifier(uid=created_uid)
133
datasource = api.datasource.get(ds_identifier)
134
print(f"Retrieved data source: {datasource['name']}")
135
136
# Alternative identification methods
137
ds_by_name = DatasourceIdentifier(name="Prometheus Production")
138
ds_by_id = DatasourceIdentifier(id="123") # Deprecated
139
140
# Health check with response model
141
health_response = api.datasource.health_check(datasource)
142
print(f"Health check - Success: {health_response.success}")
143
print(f"Status: {health_response.status}")
144
print(f"Message: {health_response.message}")
145
print(f"Duration: {health_response.duration}s")
146
147
# Convert to dictionary for logging/serialization
148
health_data = health_response.asdict_compact()
149
print(f"Compact health data: {health_data}")
150
```
151
152
### Preferences Models
153
154
User, team, and organization preference models for managing UI settings and dashboard configurations.
155
156
```python { .api }
157
@dataclass
158
class PersonalPreferences:
159
"""
160
User/team/organization preferences model.
161
162
Args:
163
homeDashboardId (Optional[int]): Home dashboard ID (deprecated)
164
homeDashboardUID (Optional[str]): Home dashboard UID (recommended)
165
locale (Optional[str]): Locale setting (e.g., "en-US", "de-DE")
166
theme (Optional[str]): UI theme ("light", "dark", "auto")
167
timezone (Optional[str]): Timezone (e.g., "browser", "utc", "America/New_York")
168
weekStart (Optional[str]): Week start day ("monday", "sunday", "saturday")
169
"""
170
homeDashboardId: Optional[int] = None
171
homeDashboardUID: Optional[str] = None
172
locale: Optional[str] = None
173
theme: Optional[str] = None
174
timezone: Optional[str] = None
175
weekStart: Optional[str] = None
176
177
def asdict(self, filter_none: bool = False) -> Dict:
178
"""
179
Convert to dictionary for API requests.
180
181
Args:
182
filter_none (bool): If True, exclude None values from output
183
184
Returns:
185
Dict: Preferences as dictionary
186
"""
187
...
188
```
189
190
**Preferences Models Usage Example:**
191
192
```python
193
from grafana_client.model import PersonalPreferences
194
195
# Create comprehensive user preferences
196
user_prefs = PersonalPreferences(
197
homeDashboardUID="home-dashboard-uid-123",
198
locale="en-US",
199
theme="dark",
200
timezone="America/New_York",
201
weekStart="monday"
202
)
203
204
# Update user preferences
205
api.user.update_preferences(user_prefs)
206
print("User preferences updated")
207
208
# Create partial preferences for team
209
team_prefs = PersonalPreferences(
210
theme="light",
211
timezone="UTC"
212
)
213
214
# Apply to team (only non-None values)
215
team_dict = team_prefs.asdict(filter_none=True)
216
api.teams.update_preferences(team_id=5, preferences=team_prefs)
217
print("Team preferences updated")
218
219
# Organization-wide default preferences
220
org_prefs = PersonalPreferences(
221
theme="auto",
222
timezone="browser",
223
weekStart="monday",
224
locale="en-US"
225
)
226
227
api.organization.update_preferences(org_prefs)
228
print("Organization default preferences set")
229
230
# Get and modify existing preferences
231
current_prefs = api.user.get_preferences()
232
print(f"Current theme: {current_prefs.theme}")
233
print(f"Current timezone: {current_prefs.timezone}")
234
235
# Partial update (patch)
236
theme_update = PersonalPreferences(theme="light")
237
api.user.patch_preferences(theme_update)
238
print("Theme updated to light")
239
```
240
241
### Exception Models
242
243
Comprehensive exception hierarchy for robust error handling with detailed error information.
244
245
```python { .api }
246
class GrafanaException(Exception):
247
"""
248
Base exception for all Grafana client errors.
249
250
Args:
251
status_code (int): HTTP status code from response
252
response: Raw HTTP response object
253
message (str): Human-readable error message
254
"""
255
def __init__(self, status_code: int, response, message: str):
256
self.status_code = status_code
257
self.response = response
258
self.message = message
259
super().__init__(message)
260
261
class GrafanaTimeoutError(GrafanaException):
262
"""
263
Raised when a request timeout occurs.
264
Inherits from GrafanaException.
265
"""
266
pass
267
268
class GrafanaServerError(GrafanaException):
269
"""
270
Raised for 5xx HTTP server errors.
271
Inherits from GrafanaException.
272
"""
273
pass
274
275
class GrafanaClientError(GrafanaException):
276
"""
277
Raised for 4xx HTTP client errors.
278
Base class for client-side errors.
279
"""
280
pass
281
282
class GrafanaBadInputError(GrafanaClientError):
283
"""
284
Raised for 400 Bad Request errors.
285
Indicates invalid input data or parameters.
286
"""
287
pass
288
289
class GrafanaUnauthorizedError(GrafanaClientError):
290
"""
291
Raised for 401 Unauthorized errors.
292
Indicates authentication failure or invalid credentials.
293
"""
294
pass
295
```
296
297
**Exception Handling Usage Example:**
298
299
```python
300
from grafana_client import (
301
GrafanaApi, TokenAuth,
302
GrafanaException, GrafanaTimeoutError, GrafanaServerError,
303
GrafanaClientError, GrafanaBadInputError, GrafanaUnauthorizedError
304
)
305
306
api = GrafanaApi(
307
auth=TokenAuth("your-token"),
308
host="grafana.example.com",
309
timeout=5.0
310
)
311
312
def robust_grafana_operation():
313
"""Example of comprehensive error handling"""
314
try:
315
# Attempt API operation
316
dashboard = api.dashboard.get_dashboard("some-dashboard-uid")
317
return dashboard
318
319
except GrafanaTimeoutError as e:
320
print(f"Operation timed out after {api.timeout}s: {e.message}")
321
print(f"Status code: {e.status_code}")
322
# Implement retry logic
323
return None
324
325
except GrafanaUnauthorizedError as e:
326
print(f"Authentication failed: {e.message}")
327
print("Check your API token and permissions")
328
# Refresh token or prompt for new credentials
329
return None
330
331
except GrafanaBadInputError as e:
332
print(f"Invalid request data: {e.message}")
333
print(f"Response details: {e.response}")
334
# Fix input data and retry
335
return None
336
337
except GrafanaClientError as e:
338
if e.status_code == 404:
339
print("Dashboard not found")
340
elif e.status_code == 403:
341
print("Access denied - insufficient permissions")
342
else:
343
print(f"Client error ({e.status_code}): {e.message}")
344
return None
345
346
except GrafanaServerError as e:
347
print(f"Grafana server error ({e.status_code}): {e.message}")
348
print("This may be a temporary server issue")
349
# Implement exponential backoff retry
350
return None
351
352
except GrafanaException as e:
353
print(f"General Grafana error ({e.status_code}): {e.message}")
354
# Log error details and handle gracefully
355
return None
356
357
except Exception as e:
358
print(f"Unexpected error: {e}")
359
# Handle non-Grafana exceptions
360
return None
361
362
# Use robust error handling
363
result = robust_grafana_operation()
364
if result:
365
print("Operation succeeded")
366
else:
367
print("Operation failed - see error messages above")
368
```
369
370
### Utility Functions and Constants
371
372
Helper functions and constants for data processing and client configuration.
373
374
```python { .api }
375
def setup_logging(level=None):
376
"""
377
Setup logging configuration for the client.
378
379
Args:
380
level: Logging level (default: logging.INFO)
381
"""
382
...
383
384
def as_bool(value: str) -> bool:
385
"""
386
Convert string representation to boolean.
387
388
Args:
389
value (str): String value to convert ("true", "false", "1", "0", etc.)
390
391
Returns:
392
bool: Converted boolean value
393
"""
394
...
395
396
def format_param_value(maybe_list):
397
"""
398
Format parameter values for HTTP requests.
399
400
Args:
401
maybe_list: Value that may be a list or single value
402
403
Returns:
404
Formatted parameter value (comma-separated if list)
405
"""
406
...
407
408
# Constants
409
DEFAULT_TIMEOUT: float = 5.0
410
DEFAULT_SESSION_POOL_SIZE: int = 10
411
```
412
413
**Utilities Usage Example:**
414
415
```python
416
from grafana_client.util import setup_logging, as_bool, format_param_value
417
from grafana_client.client import DEFAULT_TIMEOUT, DEFAULT_SESSION_POOL_SIZE
418
import logging
419
420
# Setup logging for debugging
421
setup_logging(level=logging.DEBUG)
422
print("Debug logging enabled for grafana-client")
423
424
# Boolean conversion utility
425
config_values = {
426
"enable_feature": "true",
427
"debug_mode": "1",
428
"production": "false",
429
"testing": "0"
430
}
431
432
for key, value in config_values.items():
433
bool_value = as_bool(value)
434
print(f"{key}: '{value}' -> {bool_value}")
435
436
# Parameter formatting for API requests
437
single_tag = "production"
438
multiple_tags = ["production", "monitoring", "alerts"]
439
440
formatted_single = format_param_value(single_tag)
441
formatted_multiple = format_param_value(multiple_tags)
442
443
print(f"Single tag: {formatted_single}")
444
print(f"Multiple tags: {formatted_multiple}") # "production,monitoring,alerts"
445
446
# Use constants for configuration
447
print(f"Default timeout: {DEFAULT_TIMEOUT}s")
448
print(f"Default session pool size: {DEFAULT_SESSION_POOL_SIZE}")
449
450
# Custom client configuration using constants
451
custom_api = GrafanaApi(
452
auth=TokenAuth("your-token"),
453
timeout=DEFAULT_TIMEOUT * 2, # Double the default timeout
454
session_pool_size=DEFAULT_SESSION_POOL_SIZE * 2 # Larger pool
455
)
456
```
457
458
### Complex Data Model Examples
459
460
Examples of working with complex nested data structures common in Grafana APIs.
461
462
**Dashboard Model Structure:**
463
464
```python
465
# Complete dashboard model example
466
dashboard_model = {
467
"dashboard": {
468
"id": None,
469
"uid": "custom-dashboard-uid",
470
"title": "Production Monitoring",
471
"description": "Main production monitoring dashboard",
472
"tags": ["production", "monitoring", "sre"],
473
"timezone": "UTC",
474
"editable": True,
475
"hideControls": False,
476
"schemaVersion": 30,
477
"version": 1,
478
"refresh": "30s",
479
"time": {
480
"from": "now-24h",
481
"to": "now"
482
},
483
"timepicker": {
484
"refresh_intervals": ["10s", "30s", "1m", "5m", "15m", "30m", "1h"],
485
"time_options": ["5m", "15m", "1h", "6h", "12h", "24h", "2d", "7d", "30d"]
486
},
487
"templating": {
488
"list": [
489
{
490
"name": "instance",
491
"type": "query",
492
"datasource": {"uid": "prometheus-uid"},
493
"query": "label_values(up, instance)",
494
"multi": True,
495
"includeAll": True,
496
"allValue": ".*"
497
}
498
]
499
},
500
"panels": [
501
{
502
"id": 1,
503
"title": "CPU Usage",
504
"type": "stat",
505
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 0},
506
"fieldConfig": {
507
"defaults": {
508
"unit": "percent",
509
"min": 0,
510
"max": 100,
511
"thresholds": {
512
"steps": [
513
{"color": "green", "value": None},
514
{"color": "yellow", "value": 70},
515
{"color": "red", "value": 90}
516
]
517
}
518
}
519
},
520
"targets": [
521
{
522
"expr": "100 - (avg(irate(node_cpu_seconds_total{mode=\"idle\",instance=~\"$instance\"}[5m])) * 100)",
523
"refId": "A",
524
"datasource": {"uid": "prometheus-uid"}
525
}
526
]
527
}
528
]
529
},
530
"folderId": 0,
531
"folderUID": "general",
532
"message": "Updated via API",
533
"overwrite": True
534
}
535
```
536
537
**Alert Rule Model Structure:**
538
539
```python
540
# Modern alert rule model example
541
alert_rule_model = {
542
"uid": "", # Auto-generated
543
"title": "High CPU Alert",
544
"condition": "B",
545
"data": [
546
{
547
"refId": "A",
548
"queryType": "",
549
"relativeTimeRange": {"from": 300, "to": 0},
550
"datasourceUid": "prometheus-uid",
551
"model": {
552
"expr": "avg(cpu_usage_percent) by (instance)",
553
"interval": "",
554
"refId": "A"
555
}
556
},
557
{
558
"refId": "B",
559
"queryType": "",
560
"relativeTimeRange": {"from": 0, "to": 0},
561
"datasourceUid": "__expr__",
562
"model": {
563
"expression": "A",
564
"reducer": "last",
565
"type": "reduce",
566
"refId": "B"
567
}
568
}
569
],
570
"folderUID": "alerts",
571
"ruleGroup": "Infrastructure",
572
"noDataState": "NoData",
573
"execErrState": "Alerting",
574
"for": "5m",
575
"annotations": {
576
"description": "CPU usage is above threshold",
577
"runbook_url": "https://wiki.example.com/cpu-alerts",
578
"summary": "High CPU on {{ $labels.instance }}"
579
},
580
"labels": {
581
"severity": "warning",
582
"team": "sre",
583
"component": "infrastructure"
584
}
585
}
586
```
587
588
### Type Safety Best Practices
589
590
1. **Use Data Models**: Always use provided data classes for complex payloads
591
2. **Validate Input**: Check data before sending to API
592
3. **Handle None Values**: Use `filter_none=True` for partial updates
593
4. **Exception Hierarchy**: Catch specific exception types for targeted error handling
594
5. **Type Hints**: Use type hints when extending the models
595
6. **Serialization**: Use `asdict()` methods for API compatibility
596
7. **Documentation**: Document custom data structures clearly