+++ /dev/null
-#!/usr/bin/env python3
-# Copyright 2022 Canonical Ltd.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# For those usages not covered by the Apache License, Version 2.0 please
-# contact: legal@canonical.com
-#
-# To get in touch with the maintainers, please contact:
-# osm-charmers@lists.launchpad.net
-#
-#
-# Learn more at: https://juju.is/docs/sdk
-
-"""Module to handle Grafana datasource operations."""
-
-import json
-import logging
-from dataclasses import dataclass
-
-import requests
-from requests.auth import HTTPBasicAuth
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class GrafanaConfig:
- """Values needed to make grafana API calls."""
-
- user: str
- password: str
- url: str
-
-
-@dataclass
-class DatasourceConfig:
- """Information about the datasource to create."""
-
- name: str
- url: str
-
-
-@dataclass
-class DatasourceResponse:
- """Return value used by GrafanaDataSourceHandler operations."""
-
- is_success: bool
- message: str
- results: dict
-
-
-class GrafanaDataSourceHandler:
- """Handle grafana datasource pperations."""
-
- @staticmethod
- def create_datasource(
- grafana_config: GrafanaConfig, datasource_config: DatasourceConfig
- ) -> DatasourceResponse:
- """Calls the Grafana API to create a new prometheus datasource."""
- try:
- auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
- url = grafana_config.url + "/api/datasources"
- datasource_data = {
- "name": datasource_config.name,
- "type": "prometheus",
- "url": datasource_config.url,
- "access": "proxy",
- "readOnly": False,
- "basicAuth": False,
- }
- response = requests.post(url, json=datasource_data, auth=auth)
- response_content = response.json()
- results = {"datasource-name": response_content.get("name")}
- return DatasourceResponse(
- response.ok, response_content.get("message"), results=results
- )
- except Exception as e:
- logger.debug(f"Exception processing request for creating datasource: {e}")
- return DatasourceResponse(False, str(e), results={})
-
- @staticmethod
- def list_datasources(grafana_config: GrafanaConfig) -> DatasourceResponse:
- """Calls the Grafana API to get a list of datasources."""
- try:
- auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
- url = grafana_config.url + "/api/datasources"
- response = requests.get(url, auth=auth)
- response_content = response.json()
- results = {"datasources": json.dumps(response_content)}
- message = response_content.get("message") if not response.ok else ""
- return DatasourceResponse(response.ok, message=message, results=results)
- except Exception as e:
- logger.debug(f"Exception processing request to list datasources: {e}")
- return DatasourceResponse(False, str(e), results={})
-
- @staticmethod
- def delete_datasource(
- grafana_config: GrafanaConfig, datasource_name: str
- ) -> DatasourceResponse:
- """Calls the Grafana API to delete a given datasource by name."""
- try:
- auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
- url = grafana_config.url + f"/api/datasources/name/{datasource_name}"
- response = requests.delete(url, auth=auth)
- response_content = response.json()
- return DatasourceResponse(response.ok, response_content.get("message"), results={})
- except Exception as e:
- logger.debug(f"Exception processing request for deleting datasource: {e}")
- return DatasourceResponse(False, str(e), results={})