0
# Client Management
1
2
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.
3
4
## Capabilities
5
6
### Synchronous API Client
7
8
Main synchronous client class providing access to all Grafana API elements through a unified interface.
9
10
```python { .api }
11
class GrafanaApi:
12
"""
13
Main entry point for synchronous Grafana API operations.
14
15
Args:
16
auth: Authentication credentials (TokenAuth, HeaderAuth, tuple, or AuthBase)
17
host (str): Grafana host (default: "localhost")
18
port (Optional[int]): Grafana port (optional, inferred from protocol)
19
url_path_prefix (str): URL path prefix (default: "")
20
protocol (str): Protocol (default: "http")
21
verify (bool): SSL verification (default: True)
22
timeout (float): Request timeout in seconds (default: 5.0)
23
user_agent (Optional[str]): Custom user agent string (optional)
24
organization_id (Optional[int]): Organization ID to use (optional)
25
session_pool_size (int): HTTP session pool size (default: 10)
26
"""
27
def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
28
url_path_prefix: str = "", protocol: str = "http",
29
verify: bool = True, timeout: float = 5.0,
30
user_agent: Optional[str] = None, organization_id: Optional[int] = None,
31
session_pool_size: int = 10): ...
32
33
def connect(self) -> None:
34
"""Connect to Grafana and check health"""
35
...
36
37
@property
38
def version(self) -> str:
39
"""Get Grafana version"""
40
...
41
42
@classmethod
43
def from_url(cls, url: str = None,
44
credential: Union[str, Tuple[str, str], niquests.auth.AuthBase] = None,
45
timeout: Union[float, Tuple[float, float]] = 5.0) -> 'GrafanaApi':
46
"""
47
Create GrafanaApi instance from URL.
48
49
Args:
50
url (str): Grafana URL with optional embedded credentials (default: "http://admin:admin@localhost:3000")
51
credential: Optional credential (token string, (username, password) tuple, or AuthBase instance)
52
timeout (Union[float, Tuple[float, float]]): Request timeout or (connect, read) timeouts (default: 5.0)
53
54
Returns:
55
GrafanaApi: Configured API instance
56
"""
57
...
58
59
@classmethod
60
def from_env(cls, timeout: Union[float, Tuple[float, float]] = None) -> 'GrafanaApi':
61
"""
62
Create GrafanaApi instance from environment variables.
63
64
Args:
65
timeout (Union[float, Tuple[float, float]]): Request timeout or (connect, read) timeouts (default: determined by GRAFANA_TIMEOUT or 5.0)
66
67
Returns:
68
GrafanaApi: Configured API instance from environment
69
"""
70
...
71
72
# Element API attributes (all synchronous versions)
73
@property
74
def admin(self) -> Admin: ...
75
76
@property
77
def alerting(self) -> Alerting: ...
78
79
@property
80
def alertingprovisioning(self) -> AlertingProvisioning: ...
81
82
@property
83
def annotations(self) -> Annotations: ...
84
85
@property
86
def dashboard(self) -> Dashboard: ...
87
88
@property
89
def dashboard_versions(self) -> DashboardVersions: ...
90
91
@property
92
def datasource(self) -> Datasource: ...
93
94
@property
95
def folder(self) -> Folder: ...
96
97
@property
98
def health(self) -> Health: ...
99
100
@property
101
def libraryelement(self) -> LibraryElement: ...
102
103
@property
104
def notifications(self) -> Notifications: ...
105
106
@property
107
def organization(self) -> Organization: ...
108
109
@property
110
def organizations(self) -> Organizations: ...
111
112
@property
113
def plugin(self) -> Plugin: ...
114
115
@property
116
def rbac(self) -> Rbac: ...
117
118
@property
119
def search(self) -> Search: ...
120
121
@property
122
def serviceaccount(self) -> ServiceAccount: ...
123
124
@property
125
def snapshots(self) -> Snapshots: ...
126
127
@property
128
def teams(self) -> Teams: ...
129
130
@property
131
def user(self) -> User: ...
132
133
@property
134
def users(self) -> Users: ...
135
```
136
137
**Basic Usage:**
138
139
```python
140
from grafana_client import GrafanaApi, TokenAuth
141
142
# Create client with token authentication
143
auth = TokenAuth(token="your-api-token")
144
api = GrafanaApi(
145
auth=auth,
146
host="grafana.example.com",
147
protocol="https",
148
timeout=10.0,
149
organization_id=1
150
)
151
152
# Connect and verify
153
api.connect()
154
print(f"Connected to Grafana {api.version}")
155
156
# Access API elements
157
dashboards = api.dashboard.get_dashboards_tags()
158
datasources = api.datasource.list_datasources()
159
users = api.users.search_users()
160
```
161
162
### Asynchronous API Client
163
164
Main asynchronous client class providing the same API elements as the synchronous version but with async/await support.
165
166
```python { .api }
167
class AsyncGrafanaApi:
168
"""
169
Main entry point for asynchronous Grafana API operations.
170
171
Args:
172
auth: Authentication credentials (TokenAuth, HeaderAuth, tuple, or AuthBase)
173
host (str): Grafana host (default: "localhost")
174
port (Optional[int]): Grafana port (optional, inferred from protocol)
175
url_path_prefix (str): URL path prefix (default: "")
176
protocol (str): Protocol (default: "http")
177
verify (bool): SSL verification (default: True)
178
timeout (float): Request timeout in seconds (default: 5.0)
179
user_agent (Optional[str]): Custom user agent string (optional)
180
organization_id (Optional[int]): Organization ID to use (optional)
181
"""
182
def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
183
url_path_prefix: str = "", protocol: str = "http",
184
verify: bool = True, timeout: float = 5.0,
185
user_agent: Optional[str] = None, organization_id: Optional[int] = None): ...
186
187
async def connect(self) -> None:
188
"""Connect to Grafana and check health"""
189
...
190
191
@property
192
async def version(self) -> str:
193
"""Get Grafana version"""
194
...
195
196
@classmethod
197
def from_url(cls, url: str, credential = None, timeout: float = 5.0) -> 'AsyncGrafanaApi':
198
"""Create AsyncGrafanaApi instance from URL"""
199
...
200
201
@classmethod
202
def from_env(cls, timeout: float = 5.0) -> 'AsyncGrafanaApi':
203
"""Create AsyncGrafanaApi instance from environment variables"""
204
...
205
206
# Element API attributes (all asynchronous versions)
207
@property
208
def admin(self) -> AsyncAdmin: ...
209
210
@property
211
def alerting(self) -> AsyncAlerting: ...
212
213
@property
214
def alertingprovisioning(self) -> AsyncAlertingProvisioning: ...
215
216
@property
217
def annotations(self) -> AsyncAnnotations: ...
218
219
@property
220
def dashboard(self) -> AsyncDashboard: ...
221
222
@property
223
def dashboard_versions(self) -> AsyncDashboardVersions: ...
224
225
@property
226
def datasource(self) -> AsyncDatasource: ...
227
228
@property
229
def folder(self) -> AsyncFolder: ...
230
231
@property
232
def health(self) -> AsyncHealth: ...
233
234
@property
235
def libraryelement(self) -> AsyncLibraryElement: ...
236
237
@property
238
def notifications(self) -> AsyncNotifications: ...
239
240
@property
241
def organization(self) -> AsyncOrganization: ...
242
243
@property
244
def organizations(self) -> AsyncOrganizations: ...
245
246
@property
247
def plugin(self) -> AsyncPlugin: ...
248
249
@property
250
def rbac(self) -> AsyncRbac: ...
251
252
@property
253
def search(self) -> AsyncSearch: ...
254
255
@property
256
def serviceaccount(self) -> AsyncServiceAccount: ...
257
258
@property
259
def snapshots(self) -> AsyncSnapshots: ...
260
261
@property
262
def teams(self) -> AsyncTeams: ...
263
264
@property
265
def user(self) -> AsyncUser: ...
266
267
@property
268
def users(self) -> AsyncUsers: ...
269
```
270
271
**Basic Async Usage:**
272
273
```python
274
import asyncio
275
from grafana_client import AsyncGrafanaApi, TokenAuth
276
277
async def main():
278
# Create async client
279
auth = TokenAuth(token="your-api-token")
280
api = AsyncGrafanaApi(
281
auth=auth,
282
host="grafana.example.com",
283
protocol="https",
284
timeout=10.0
285
)
286
287
# Connect and verify
288
await api.connect()
289
version = await api.version
290
print(f"Connected to Grafana {version}")
291
292
# Access async API elements
293
dashboards = await api.dashboard.get_dashboards_tags()
294
datasources = await api.datasource.list_datasources()
295
users = await api.users.search_users()
296
297
# Run async code
298
asyncio.run(main())
299
```
300
301
### Low-Level HTTP Clients
302
303
Direct HTTP client classes for advanced use cases requiring fine-grained control over requests.
304
305
```python { .api }
306
class GrafanaClient:
307
"""
308
Low-level synchronous HTTP client for Grafana API.
309
310
Args:
311
auth: Authentication credentials
312
host (str): Grafana host (default: "localhost")
313
port (Optional[int]): Grafana port (optional)
314
url_path_prefix (str): URL path prefix (default: "")
315
protocol (str): Protocol (default: "http")
316
verify (bool): SSL verification (default: True)
317
timeout (float): Request timeout
318
user_agent (Optional[str]): Custom user agent string
319
organization_id (Optional[int]): Organization ID to use
320
session_pool_size (int): HTTP session pool size
321
"""
322
def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
323
url_path_prefix: str = "", protocol: str = "http",
324
verify: bool = True, timeout: float = None,
325
user_agent: Optional[str] = None, organization_id: Optional[int] = None,
326
session_pool_size: int = 10): ...
327
328
class AsyncGrafanaClient:
329
"""
330
Low-level asynchronous HTTP client for Grafana API.
331
332
Same constructor parameters as GrafanaClient.
333
"""
334
def __init__(self, auth, host: str = "localhost", port: Optional[int] = None,
335
url_path_prefix: str = "", protocol: str = "http",
336
verify: bool = True, timeout: float = None,
337
user_agent: Optional[str] = None, organization_id: Optional[int] = None): ...
338
```
339
340
### Factory Methods
341
342
Convenient methods for creating client instances from URLs or environment variables.
343
344
**URL-based Factory:**
345
346
```python
347
# URL with embedded credentials
348
api = GrafanaApi.from_url("https://admin:password@grafana.example.com:3000/grafana")
349
350
# URL with separate credential
351
api = GrafanaApi.from_url(
352
url="https://grafana.example.com:3000",
353
credential="your-api-token",
354
timeout=15.0
355
)
356
357
# Custom path prefix
358
api = GrafanaApi.from_url("https://grafana.example.com/custom-path")
359
```
360
361
**Environment-based Factory:**
362
363
Set environment variables:
364
```bash
365
export GRAFANA_HOST=grafana.example.com
366
export GRAFANA_PROTOCOL=https
367
export GRAFANA_TOKEN=your-api-token
368
export GRAFANA_ORGANIZATION_ID=1
369
export GRAFANA_TIMEOUT=10.0
370
export GRAFANA_VERIFY_SSL=true
371
```
372
373
Create client:
374
```python
375
# Automatically configured from environment
376
api = GrafanaApi.from_env()
377
378
# With custom timeout
379
api = GrafanaApi.from_env(timeout=20.0)
380
```
381
382
### Connection Management
383
384
**Health Checking:**
385
386
```python
387
from grafana_client import GrafanaApi, GrafanaException
388
389
try:
390
api = GrafanaApi(auth=auth, host="grafana.example.com")
391
api.connect() # Validates connection and health
392
print("Grafana is healthy")
393
except GrafanaException as e:
394
print(f"Connection failed: {e.message}")
395
```
396
397
**Timeout Configuration:**
398
399
```python
400
# Global timeout for all requests
401
api = GrafanaApi(auth=auth, timeout=30.0)
402
403
# Different timeouts for different operations
404
api.connect() # Uses global timeout
405
406
# Individual request timeouts handled by underlying HTTP client
407
```
408
409
**SSL Configuration:**
410
411
```python
412
# Disable SSL verification (not recommended for production)
413
api = GrafanaApi(auth=auth, protocol="https", verify=False)
414
415
# Custom SSL configuration through requests
416
import requests
417
session = requests.Session()
418
session.verify = "/path/to/ca-bundle.crt"
419
# Use custom session through low-level client if needed
420
```
421
422
### Organization Context
423
424
**Setting Organization Context:**
425
426
```python
427
# Set organization during client creation
428
api = GrafanaApi(auth=auth, organization_id=5)
429
430
# Switch organization context at runtime
431
api.organizations.switch_organization(organization_id=3)
432
433
# Get current organization
434
current_org = api.organization.get_current_organization()
435
print(f"Current organization: {current_org['name']}")
436
```
437
438
### Session Management
439
440
**HTTP Session Pool Configuration:**
441
442
```python
443
# Configure connection pooling for high-throughput scenarios
444
api = GrafanaApi(
445
auth=auth,
446
session_pool_size=50, # Larger pool for concurrent requests
447
timeout=5.0
448
)
449
450
# Async API doesn't use session pool (uses aiohttp)
451
async_api = AsyncGrafanaApi(auth=auth, timeout=5.0)
452
```
453
454
### Error Handling
455
456
Connection and client-level errors:
457
458
```python
459
from grafana_client import (
460
GrafanaApi, GrafanaException, GrafanaTimeoutError,
461
GrafanaServerError, GrafanaClientError
462
)
463
464
try:
465
api = GrafanaApi(auth=auth, host="invalid-host", timeout=2.0)
466
api.connect()
467
except GrafanaTimeoutError as e:
468
print(f"Connection timeout: {e}")
469
except GrafanaServerError as e:
470
print(f"Server error ({e.status_code}): {e.message}")
471
except GrafanaClientError as e:
472
print(f"Client error ({e.status_code}): {e.message}")
473
except GrafanaException as e:
474
print(f"General error: {e}")
475
```
476
477
### Performance Considerations
478
479
**Synchronous vs Asynchronous:**
480
481
```python
482
# Use sync API for simple scripts and single-threaded applications
483
api = GrafanaApi(auth=auth)
484
result = api.dashboard.get_dashboard("dashboard-uid")
485
486
# Use async API for concurrent operations and high-throughput scenarios
487
async def fetch_multiple_dashboards():
488
api = AsyncGrafanaApi(auth=auth)
489
tasks = [
490
api.dashboard.get_dashboard(uid)
491
for uid in dashboard_uids
492
]
493
results = await asyncio.gather(*tasks)
494
return results
495
```
496
497
**Connection Reuse:**
498
499
```python
500
# Reuse the same client instance for multiple operations
501
api = GrafanaApi(auth=auth)
502
api.connect() # Establish connection once
503
504
# Multiple operations reuse the connection
505
dashboard1 = api.dashboard.get_dashboard("uid1")
506
dashboard2 = api.dashboard.get_dashboard("uid2")
507
datasources = api.datasource.list_datasources()
508
```
509
510
## Constants
511
512
```python { .api }
513
DEFAULT_TIMEOUT: float = 5.0
514
DEFAULT_SESSION_POOL_SIZE: int = 10
515
```