tessl/pypi-grafana-client
A client library for accessing the Grafana HTTP API, written in Python
—
Pending
admin-and-rbac.mddocs/
Admin and RBAC
Administrative functions, user management, role-based access control, service accounts, and system-level operations with comprehensive permission management. This covers both administrative operations and the fine-grained RBAC system introduced in Grafana Enterprise.
Capabilities
Administrative Operations
System-level administrative functions for managing users, settings, and global configurations.
def settings(self):
"""
Get admin settings and configuration.
Returns:
dict: System settings and configuration parameters
"""
...
def stats(self):
"""
Get admin statistics about the Grafana instance.
Returns:
dict: Usage statistics, user counts, dashboard counts, etc.
"""
...
def create_user(self, user: dict):
"""
Create a new user (admin only).
Args:
user (dict): User data (name, email, login, password, orgId)
Returns:
dict: Created user with ID and metadata
"""
...
def change_user_password(self, user_id: int, password: str):
"""
Change user's password (admin only).
Args:
user_id (int): User ID
password (str): New password
Returns:
dict: Password change result
"""
...
def change_user_permissions(self, user_id: int, is_grafana_admin: bool):
"""
Change user's admin permissions (admin only).
Args:
user_id (int): User ID
is_grafana_admin (bool): Grant/revoke admin permissions
Returns:
dict: Permission change result
"""
...
def delete_user(self, user_id: int):
"""
Delete user (admin only).
Args:
user_id (int): User ID to delete
Returns:
dict: Deletion result
"""
...
def pause_all_alerts(self, pause: bool):
"""
Pause or unpause all alerts globally (admin only).
Args:
pause (bool): True to pause alerts, False to unpause
Returns:
dict: Pause operation result
"""
...
def set_user_enabled(self, user_id: int, enabled: bool):
"""
Enable or disable user account (admin only).
Args:
user_id (int): User ID
enabled (bool): True to enable, False to disable
Returns:
dict: User status change result
"""
...Administrative Operations Usage Example:
from grafana_client import GrafanaApi, TokenAuth
# Must use admin token/credentials
api = GrafanaApi(auth=TokenAuth("admin-token"), host="grafana.example.com")
# Get system statistics
stats = api.admin.stats()
print(f"Total users: {stats.get('users', 0)}")
print(f"Total dashboards: {stats.get('dashboards', 0)}")
print(f"Total organizations: {stats.get('orgs', 0)}")
# Get system settings (be careful with sensitive data)
settings = api.admin.settings()
print(f"Grafana version: {settings.get('buildInfo', {}).get('version', 'unknown')}")
# Create new user
new_user = {
"name": "John Doe",
"email": "john.doe@example.com",
"login": "johndoe",
"password": "secure-password-123",
"orgId": 1
}
created_user = api.admin.create_user(new_user)
user_id = created_user['id']
print(f"Created user: {user_id}")
# Grant admin permissions
api.admin.change_user_permissions(user_id, is_grafana_admin=True)
print(f"Granted admin permissions to user {user_id}")
# Change user password
api.admin.change_user_password(user_id, "new-secure-password-456")
print("Password changed")
# Disable user temporarily
api.admin.set_user_enabled(user_id, enabled=False)
print("User disabled")
# Enable user
api.admin.set_user_enabled(user_id, enabled=True)
print("User enabled")
# Pause all alerts (emergency operation)
api.admin.pause_all_alerts(pause=True)
print("All alerts paused")
# Unpause alerts
api.admin.pause_all_alerts(pause=False)
print("All alerts unpaused")Role-Based Access Control (RBAC)
Fine-grained permission management system for controlling access to resources (Grafana Enterprise feature).
def get_rbac_roles_all(self):
"""
Get all RBAC roles available in the system.
Returns:
list: List of all RBAC roles with permissions
"""
...
def add_rbac_role_team(self, team_id: int, role_uid: str):
"""
Add RBAC role to team.
Args:
team_id (int): Team ID
role_uid (str): RBAC role UID to assign
Returns:
dict: Role assignment result
"""
...
def remove_rbac_role_team(self, team_id: int, role_uid: str):
"""
Remove RBAC role from team.
Args:
team_id (int): Team ID
role_uid (str): RBAC role UID to remove
Returns:
dict: Role removal result
"""
...
def get_rbac_datasources(self, datasource_uid: str):
"""
Get RBAC permissions for data source.
Args:
datasource_uid (str): Data source UID
Returns:
dict: Data source RBAC permissions
"""
...
def set_rbac_datasources_teams(self, datasource_uid: str, team_id: int, permission: str):
"""
Set team RBAC permissions for data source.
Args:
datasource_uid (str): Data source UID
team_id (int): Team ID
permission (str): Permission level ("Query", "Edit", "Admin")
Returns:
dict: Permission assignment result
"""
...
def set_rbac_datasources_builtin_roles(self, datasource_uid: str, builtin_role: str, permission: str):
"""
Set built-in role RBAC permissions for data source.
Args:
datasource_uid (str): Data source UID
builtin_role (str): Built-in role ("Viewer", "Editor", "Admin")
permission (str): Permission level ("Query", "Edit", "Admin")
Returns:
dict: Permission assignment result
"""
...RBAC Usage Example:
# Get all available RBAC roles
rbac_roles = api.rbac.get_rbac_roles_all()
print("Available RBAC roles:")
for role in rbac_roles:
print(f"- {role['name']} ({role['uid']}): {role.get('description', 'No description')}")
# Assign specific role to team
custom_role_uid = "custom:reports.creator"
team_id = 5
try:
api.rbac.add_rbac_role_team(team_id, custom_role_uid)
print(f"Assigned role {custom_role_uid} to team {team_id}")
except Exception as e:
print(f"Failed to assign role: {e}")
# Get data source permissions
datasource_uid = "prometheus-main"
ds_permissions = api.rbac.get_rbac_datasources(datasource_uid)
print(f"Data source permissions: {ds_permissions}")
# Set team permissions for data source
api.rbac.set_rbac_datasources_teams(
datasource_uid=datasource_uid,
team_id=team_id,
permission="Query" # Query, Edit, or Admin
)
print(f"Set Query permission for team {team_id} on data source {datasource_uid}")
# Set built-in role permissions
api.rbac.set_rbac_datasources_builtin_roles(
datasource_uid=datasource_uid,
builtin_role="Editor",
permission="Query"
)
print("Set Query permission for Editor role on data source")
# Remove role from team
api.rbac.remove_rbac_role_team(team_id, custom_role_uid)
print(f"Removed role {custom_role_uid} from team {team_id}")Service Account Management
Managing service accounts for programmatic access and automation.
def get(self, service_account_id: int):
"""
Get service account by ID.
Args:
service_account_id (int): Service account ID
Returns:
dict: Service account information
"""
...
def create(self, service_account: dict):
"""
Create new service account.
Args:
service_account (dict): Service account configuration (name, role)
Returns:
dict: Created service account with ID
"""
...
def update(self, service_account_id: int, service_account: dict):
"""
Update service account.
Args:
service_account_id (int): Service account ID
service_account (dict): Updated configuration
Returns:
dict: Update result
"""
...
def delete(self, service_account_id: int):
"""
Delete service account.
Args:
service_account_id (int): Service account ID to delete
Returns:
dict: Deletion result
"""
...
def search(self, query: Optional[str] = None, page: int = 1, perpage: int = 1000):
"""
Search service accounts.
Args:
query (Optional[str]): Search query
page (int): Page number (default: 1)
perpage (int): Results per page (default: 1000)
Returns:
list: List of service accounts matching criteria
"""
...
def search_one(self, service_account_name: str):
"""
Search for single service account by name.
Args:
service_account_name (str): Service account name
Returns:
dict: Service account if found, None otherwise
"""
...
def get_tokens(self, service_account_id: int):
"""
Get tokens for service account.
Args:
service_account_id (int): Service account ID
Returns:
list: List of tokens associated with service account
"""
...
def create_token(self, service_account_id: int, content: dict):
"""
Create token for service account.
Args:
service_account_id (int): Service account ID
content (dict): Token configuration (name, secondsToLive)
Returns:
dict: Created token with key (only returned once)
"""
...
def delete_token(self, service_account_id: int, service_account_token_id: int):
"""
Delete service account token.
Args:
service_account_id (int): Service account ID
service_account_token_id (int): Token ID to delete
Returns:
dict: Deletion result
"""
...Service Account Management Usage Example:
# Create service account for automation
automation_sa = {
"name": "automation-service",
"role": "Editor" # Viewer, Editor, or Admin
}
created_sa = api.serviceaccount.create(automation_sa)
sa_id = created_sa['id']
print(f"Created service account: {sa_id}")
# Create token for service account
token_config = {
"name": "automation-token-prod",
"secondsToLive": 86400 * 365 # 1 year
}
token_result = api.serviceaccount.create_token(sa_id, token_config)
token_key = token_result['key'] # Store this securely - only shown once
print(f"Created token: {token_result['name']} (expires: {token_result['expiration']})")
# Search service accounts
all_service_accounts = api.serviceaccount.search()
print(f"Total service accounts: {len(all_service_accounts)}")
# Search for specific service account
automation_accounts = api.serviceaccount.search(query="automation")
for sa in automation_accounts:
print(f"Service account: {sa['name']} (ID: {sa['id']}, Role: {sa['role']})")
# Find service account by name
specific_sa = api.serviceaccount.search_one("automation-service")
if specific_sa:
print(f"Found service account: {specific_sa['name']}")
# Get service account details
sa_details = api.serviceaccount.get(sa_id)
print(f"Service account details: {sa_details}")
# List tokens for service account
tokens = api.serviceaccount.get_tokens(sa_id)
print(f"Service account has {len(tokens)} tokens")
for token in tokens:
print(f"Token: {token['name']} (ID: {token['id']}, Expires: {token['expiration']})")
# Update service account
api.serviceaccount.update(sa_id, {
"name": "automation-service-updated",
"role": "Admin"
})
print("Service account updated")
# Delete token (when rotating)
if tokens:
token_to_delete = tokens[0]['id']
api.serviceaccount.delete_token(sa_id, token_to_delete)
print(f"Deleted token {token_to_delete}")
# Create new token for rotation
new_token = api.serviceaccount.create_token(sa_id, {
"name": "automation-token-rotated",
"secondsToLive": 86400 * 90 # 90 days
})
print(f"Created replacement token: {new_token['name']}")Health and System Monitoring
System health checks and monitoring capabilities for administrative oversight.
# Health check (via Health API)
def check(self):
"""
Check Grafana system health.
Returns:
dict: Health status information
"""
...System Health Usage Example:
# Check system health
try:
health_status = api.health.check()
print(f"System health: {health_status}")
if health_status.get('database') == 'ok':
print("Database connection is healthy")
else:
print("Database connection issues detected")
except Exception as e:
print(f"Health check failed: {e}")
# Combined admin monitoring
def admin_health_check():
"""Comprehensive admin health check"""
try:
# System stats
stats = api.admin.stats()
print(f"System load - Users: {stats.get('users', 0)}, "
f"Dashboards: {stats.get('dashboards', 0)}, "
f"Data sources: {stats.get('datasources', 0)}")
# Health status
health = api.health.check()
print(f"Health status: {health}")
# Service accounts audit
service_accounts = api.serviceaccount.search()
print(f"Active service accounts: {len(service_accounts)}")
return True
except Exception as e:
print(f"Admin health check failed: {e}")
return False
# Run comprehensive check
admin_health_check()Bulk Administrative Operations
Performing administrative operations at scale:
# Bulk user management
def bulk_user_operations():
"""Example of bulk user management operations"""
# Get all users for audit
all_users = api.users.search_users()
print(f"Total users to audit: {len(all_users)}")
admin_users = []
inactive_users = []
for user in all_users:
user_details = api.users.get_user(user['id'])
# Check if user is admin
if user_details.get('isGrafanaAdmin', False):
admin_users.append(user_details)
# Check last login (if available in user details)
# This would depend on your Grafana version and configuration
if not user_details.get('lastSeenAt'):
inactive_users.append(user_details)
print(f"Admin users: {len(admin_users)}")
print(f"Potentially inactive users: {len(inactive_users)}")
# Disable inactive users (with caution)
for user in inactive_users[:5]: # Limit to first 5 for safety
try:
api.admin.set_user_enabled(user['id'], enabled=False)
print(f"Disabled inactive user: {user['login']}")
except Exception as e:
print(f"Failed to disable user {user['login']}: {e}")
# Bulk RBAC operations
def bulk_rbac_setup():
"""Set up RBAC permissions for multiple teams and data sources"""
# Get all teams
teams = api.teams.search_teams()
# Get all data sources
datasources = api.datasource.list_datasources()
# Define permission mapping
team_permissions = {
"DevOps": "Admin",
"Development": "Edit",
"QA": "Query",
"Support": "Query"
}
for team in teams:
team_name = team['name']
if team_name in team_permissions:
permission = team_permissions[team_name]
# Apply permission to all data sources
for ds in datasources:
try:
api.rbac.set_rbac_datasources_teams(
datasource_uid=ds['uid'],
team_id=team['id'],
permission=permission
)
print(f"Set {permission} permission for team {team_name} on {ds['name']}")
except Exception as e:
print(f"Failed to set permission: {e}")
# Run bulk operations (with caution)
# bulk_user_operations()
# bulk_rbac_setup()Error Handling
Administrative operation error handling:
from grafana_client import GrafanaClientError, GrafanaUnauthorizedError
try:
# Administrative operation requiring admin privileges
stats = api.admin.stats()
except GrafanaUnauthorizedError as e:
print(f"Admin access required: {e.message}")
print("Current user does not have admin privileges")
except GrafanaClientError as e:
if e.status_code == 403:
print("Forbidden: Insufficient permissions for admin operation")
else:
print(f"Admin operation failed: {e.message}")
# Service account operation error handling
try:
# Create service account with invalid data
invalid_sa = {
"name": "", # Empty name
"role": "InvalidRole" # Invalid role
}
api.serviceaccount.create(invalid_sa)
except Exception as e:
print(f"Service account creation failed: {e}")Async Admin and RBAC Operations
All admin and RBAC operations support async versions:
import asyncio
from grafana_client import AsyncGrafanaApi, TokenAuth
async def async_admin_operations():
api = AsyncGrafanaApi(auth=TokenAuth("admin-token"), host="grafana.example.com")
# Concurrent admin operations
admin_tasks = [
api.admin.stats(),
api.admin.settings(),
api.rbac.get_rbac_roles_all(),
api.serviceaccount.search()
]
stats, settings, rbac_roles, service_accounts = await asyncio.gather(*admin_tasks)
print(f"System stats: {len(stats)} metrics")
print(f"RBAC roles: {len(rbac_roles)}")
print(f"Service accounts: {len(service_accounts)}")
# Concurrent service account token management
token_tasks = []
for sa in service_accounts[:3]: # First 3 service accounts
task = api.serviceaccount.get_tokens(sa['id'])
token_tasks.append(task)
token_results = await asyncio.gather(*token_tasks, return_exceptions=True)
for i, tokens in enumerate(token_results):
if isinstance(tokens, Exception):
print(f"Failed to get tokens for SA {i}: {tokens}")
else:
print(f"Service account {i} has {len(tokens)} tokens")
asyncio.run(async_admin_operations())Best Practices
- Security First: Always use principle of least privilege for admin operations
- Audit Regularly: Monitor admin actions and service account usage
- Token Rotation: Implement regular rotation for service account tokens
- RBAC Planning: Design RBAC roles and permissions systematically
- Backup First: Always backup before bulk administrative changes
- Error Handling: Implement robust error handling for admin operations
- Monitoring: Monitor system health and performance metrics
- Documentation: Document all administrative procedures and RBAC policies
Install with Tessl CLI
npx tessl i tessl/pypi-grafana-client@5.0.1