Add cw 22 support

docs/content/grafana_api/api.md

@@ -34,7 +34,8 @@ The class includes all necessary methods to make API calls to the Grafana API en
 ```python
 def call_the_api(api_call: str,
                  method: RequestsMethods = RequestsMethods.GET,
-                 json_complete: str = None) -> any
+                 json_complete: str = None,
+                 timeout: float = None) -> any
 ```
 
 The method execute a defined API call against the Grafana endpoints
@@ -44,6 +45,7 @@ The method execute a defined API call against the Grafana endpoints
 - `api_call` _str_ - Specify the API call endpoint
 - `method` _RequestsMethods_ - Specify the used method (default GET)
 - `json_complete` _str_ - Specify the inserted JSON as string
+- `timeout` _float_ - Specify the timeout for the corresponding API call
 
 
 **Raises**:

docs/content/grafana_api/model.md

@@ -15,6 +15,8 @@
   * [PlaylistObject](#grafana_api.model.PlaylistObject)
   * [PlaylistItemObject](#grafana_api.model.PlaylistItemObject)
   * [TeamObject](#grafana_api.model.TeamObject)
+  * [QueryDatasourceObject](#grafana_api.model.QueryDatasourceObject)
+  * [QueryObject](#grafana_api.model.QueryObject)
 
 <a id="grafana_api.model"></a>
 
@@ -259,3 +261,35 @@ The class includes all necessary variables to generate a team object that is nec
 - `email` _str_ - Specify the email of the team
 - `org_id` _int_ - Specify the org_id of the team
 
+<a id="grafana_api.model.QueryDatasourceObject"></a>
+
+## QueryDatasourceObject Objects
+
+```python
+class QueryDatasourceObject(NamedTuple)
+```
+
+The class includes all necessary variables to generate a query datasource object that is necessary to create a query history object
+
+**Arguments**:
+
+- `type` _str_ - Specify the type of the datasource query
+- `uid` _str_ - Specify the uid of the datasource query
+
+<a id="grafana_api.model.QueryObject"></a>
+
+## QueryObject Objects
+
+```python
+class QueryObject(NamedTuple)
+```
+
+The class includes all necessary variables to generate a query object that is necessary to create a query history
+
+**Arguments**:
+
+- `ref_id` _str_ - Specify the ref_id of the query history
+- `key` _str_ - Specify the key of the query history
+- `scenario_id` _str_ - Specify the scenario_id of the query history
+- `datasource` _QueryDatasourceObject_ - Specify the datasource of the type QueryDatasourceObject
+

src/grafana_api/api.py

@@ -1,4 +1,5 @@
 import logging
+import json
 
 import requests
 
@@ -23,13 +24,15 @@ def call_the_api(
         api_call: str,
         method: RequestsMethods = RequestsMethods.GET,
         json_complete: str = None,
+        timeout: float = None,
     ) -> any:
         """The method execute a defined API call against the Grafana endpoints
 
         Args:
             api_call (str): Specify the API call endpoint
             method (RequestsMethods): Specify the used method (default GET)
             json_complete (str): Specify the inserted JSON as string
+            timeout (float): Specify the timeout for the corresponding API call
 
         Raises:
             Exception: Unspecified error by executing the API call
@@ -58,35 +61,50 @@ def call_the_api(
         try:
             if method.value == RequestsMethods.GET.value:
                 return Api.__check_the_api_call_response(
-                    requests.get(api_url, headers=headers)
+                    requests.get(api_url, headers=headers, timeout=timeout)
                 )
             elif method.value == RequestsMethods.PUT.value:
                 if json_complete is not None:
                     return Api.__check_the_api_call_response(
-                        requests.put(api_url, data=json_complete, headers=headers)
+                        requests.put(
+                            api_url,
+                            data=json_complete,
+                            headers=headers,
+                            timeout=timeout,
+                        )
                     )
                 else:
                     logging.error("Please define the json_complete.")
                     raise Exception
             elif method.value == RequestsMethods.POST.value:
                 if json_complete is not None:
                     return Api.__check_the_api_call_response(
-                        requests.post(api_url, data=json_complete, headers=headers)
+                        requests.post(
+                            api_url,
+                            data=json_complete,
+                            headers=headers,
+                            timeout=timeout,
+                        )
                     )
                 else:
                     logging.error("Please define the json_complete.")
                     raise Exception
             elif method.value == RequestsMethods.PATCH.value:
                 if json_complete is not None:
                     return Api.__check_the_api_call_response(
-                        requests.patch(api_url, data=json_complete, headers=headers)
+                        requests.patch(
+                            api_url,
+                            data=json_complete,
+                            headers=headers,
+                            timeout=timeout,
+                        )
                     )
                 else:
                     logging.error("Please define the json_complete.")
                     raise Exception
             elif method.value == RequestsMethods.DELETE.value:
                 return Api.__check_the_api_call_response(
-                    requests.delete(api_url, headers=headers)
+                    requests.delete(api_url, headers=headers, timeout=timeout)
                 )
             else:
                 logging.error("Please define a valid method.")
@@ -108,12 +126,30 @@ def __check_the_api_call_response(response: any = None) -> any:
             api_call (any): Returns the value of the api call
         """
 
-        if len(response.text) != 0 and type(response.json()) == dict:
-            if (
-                "message" in response.json().keys()
-                and response.json()["message"] in ERROR_MESSAGES
-            ):
-                logging.error(response.json()["message"])
-                raise requests.exceptions.ConnectionError
+        if Api.__check_if_valid_json(response.text):
+            if len(response.text) != 0 and type(response.json()) == dict:
+                if (
+                    "message" in response.json().keys()
+                    and response.json()["message"] in ERROR_MESSAGES
+                ):
+                    logging.error(response.json()["message"])
+                    raise requests.exceptions.ConnectionError
 
         return response
+
+    @staticmethod
+    def __check_if_valid_json(response: any) -> bool:
+        """The method includes a functionality to check if the response json is valid
+
+        Args:
+            response (any): Specify the inserted response json
+
+        Returns:
+            api_call (bool): Returns if the json is valid or not
+        """
+
+        try:
+            json.loads(response)
+        except ValueError:
+            return False
+        return True

src/grafana_api/licensing.py

@@ -0,0 +1,107 @@
+import json
+from requests import Response
+import logging
+
+from .model import (
+    APIModel,
+    APIEndpoints,
+    RequestsMethods,
+)
+from .api import Api
+
+
+class Licensing:
+    """The class includes all necessary methods to access the Grafana licensing API endpoints. Be aware that the functionality is a Grafana ENTERPRISE v7.4+ feature
+
+    HINT: Note Grafana Enterprise API need required permissions if fine-grained access control is enabled
+
+    Args:
+        grafana_api_model (APIModel): Inject a Grafana API model object that includes all necessary values and information
+
+    Attributes:
+        grafana_api_model (APIModel): This is where we store the grafana_api_model
+    """
+
+    def __init__(self, grafana_api_model: APIModel):
+        self.grafana_api_model = grafana_api_model
+
+    def check_license_availability(self):
+        """The method includes a functionality to checks if a valid license is available
+
+        Required Permissions:
+            Action: licensing:read
+            Scope: N/A
+
+        Raises:
+            Exception: Unspecified error by executing the API call
+
+        Returns:
+            api_call (bool): Returns the result if the license is available or not
+        """
+
+        api_call: Response = Api(self.grafana_api_model).call_the_api(
+            f"{APIEndpoints.LICENSING.value}/check",
+        )
+
+        if api_call.status_code != 200:
+            logging.error(f"Check the error: {api_call}.")
+            raise Exception
+        else:
+            return json.loads(str(api_call.text))
+
+    def manually_force_license_refresh(self):
+        """The method includes a functionality to manually ask license issuer for a new token
+
+        Required Permissions:
+            Action: licensing:update
+            Scope: N/A
+
+        Raises:
+            Exception: Unspecified error by executing the API call
+
+        Returns:
+            api_call (dict): Returns the result of license refresh call
+        """
+
+        api_call: dict = (
+            Api(self.grafana_api_model)
+            .call_the_api(
+                f"{APIEndpoints.LICENSING.value}/token/renew",
+                RequestsMethods.POST,
+                json.dumps({}),
+            )
+            .json()
+        )
+
+        if api_call == dict() or api_call.get("jti") is None:
+            logging.error(f"Check the error: {api_call}.")
+            raise Exception
+        else:
+            return api_call
+
+    def remove_license_from_dashboard(self):
+        """The method includes a functionality to removes the license stored in the Grafana database
+
+        Required Permissions:
+            Action: licensing:delete
+            Scope: N/A
+
+        Raises:
+            Exception: Unspecified error by executing the API call
+
+        Returns:
+            api_call (dict): Returns the result of license refresh call
+        """
+
+        api_call: Response = Api(self.grafana_api_model).call_the_api(
+            f"{APIEndpoints.LICENSING.value}/token",
+            RequestsMethods.DELETE,
+        )
+
+        if api_call.status_code != 200:
+            logging.error(f"Check the error: {api_call}.")
+            raise Exception
+        else:
+            logging.info(
+                "You successfully removed the corresponding license from the database."
+            )

src/grafana_api/model.py

@@ -31,6 +31,11 @@ class APIEndpoints(Enum):
     DASHBOARD_SNAPSHOTS = f"{api_prefix}/dashboard/snapshots"
     PLAYLISTS = f"{api_prefix}/playlists"
     TEAMS = f"{api_prefix}/teams"
+    QUERY_HISTORY = f"{api_prefix}/query-history"
+    REPORTING = f"{api_prefix}/reports/email"
+    LICENSING = f"{api_prefix}/licensing"
+    FRONTEND = f"{api_prefix}/frontend"
+    LOGIN = f"{api_prefix}/login"
 
 
 class RequestsMethods(Enum):
@@ -264,3 +269,31 @@ class TeamObject(NamedTuple):
     name: str
     email: str
     org_id: int
+
+
+class QueryDatasourceObject(NamedTuple):
+    """The class includes all necessary variables to generate a query datasource object that is necessary to create a query history object
+
+    Args:
+        type (str): Specify the type of the datasource query
+        uid (str): Specify the uid of the datasource query
+    """
+
+    type: str
+    uid: str
+
+
+class QueryObject(NamedTuple):
+    """The class includes all necessary variables to generate a query object that is necessary to create a query history
+
+    Args:
+        ref_id (str): Specify the ref_id of the query history
+        key (str): Specify the key of the query history
+        scenario_id (str): Specify the scenario_id of the query history
+        datasource (QueryDatasourceObject): Specify the datasource of the type QueryDatasourceObject
+    """
+
+    ref_id: str
+    key: str
+    scenario_id: str
+    datasource: QueryDatasourceObject