Issue 118

docs/content/grafana_api/datasource.md

@@ -30,6 +30,9 @@
     * [disable\_datasource\_cache](#datasource.DatasourceQueryResourceCaching.disable_datasource_cache)
     * [clean\_datasource\_cache](#datasource.DatasourceQueryResourceCaching.clean_datasource_cache)
     * [update\_datasource\_cache](#datasource.DatasourceQueryResourceCaching.update_datasource_cache)
+  * [DatasourceLabelBasedAccessControl](#datasource.DatasourceLabelBasedAccessControl)
+    * [get\_lbac\_rules\_for\_datasource](#datasource.DatasourceLabelBasedAccessControl.get_lbac_rules_for_datasource)
+    * [update\_lbac\_rules\_for\_datasource](#datasource.DatasourceLabelBasedAccessControl.update_lbac_rules_for_datasource)
 
 <a id="datasource"></a>
 
@@ -852,3 +855,80 @@ The method includes a functionality to update the datasource cache specified by
 
 - `api_call` _dict_ - Returns a datasource
 
+<a id="datasource.DatasourceLabelBasedAccessControl"></a>
+
+## DatasourceLabelBasedAccessControl Objects
+
+```python
+class DatasourceLabelBasedAccessControl()
+```
+
+The class includes all necessary methods to access the Grafana datasource label based access control for teams API endpoints. It's required that the API token got the corresponding datasource access rights. Please check the used methods docstring for the necessary access rights. The functionality is a Grafana Cloud feature. Only cloud Loki data sources are supported
+
+**Arguments**:
+
+- `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
+
+<a id="datasource.DatasourceLabelBasedAccessControl.get_lbac_rules_for_datasource"></a>
+
+#### get\_lbac\_rules\_for\_datasource
+
+```python
+def get_lbac_rules_for_datasource(uid: str) -> list
+```
+
+The method includes a functionality to get all datasource label based access control rules for team specified by the datasource uid
+
+**Arguments**:
+
+- `uid` _str_ - Specify the uid of the datasource
+
+  Required Permissions:
+- `Action` - datasources:read
+- `Scope` - [datasources:*, datasources:uid:*, datasources:uid:<id>]
+
+
+**Raises**:
+
+- `ValueError` - Missed specifying a necessary value
+- `Exception` - Unspecified error by executing the API call
+
+
+**Returns**:
+
+- `api_call` _list_ - Returns all LBAC rules
+
+<a id="datasource.DatasourceLabelBasedAccessControl.update_lbac_rules_for_datasource"></a>
+
+#### update\_lbac\_rules\_for\_datasource
+
+```python
+def update_lbac_rules_for_datasource(uid: str) -> dict
+```
+
+The method includes a functionality to enable the datasource cache specified by the datasource uid
+
+**Arguments**:
+
+- `uid` _str_ - Specify the uid of the datasource
+
+  Required Permissions:
+- `Action` - datasources:write, datasources.permissions:write
+- `Scope` - [datasources:*, datasources:uid:*, datasources:uid:<id>]
+
+
+**Raises**:
+
+- `ValueError` - Missed specifying a necessary value
+- `Exception` - Unspecified error by executing the API call
+
+
+**Returns**:
+
+- `api_call` _dict_ - Returns a datasource
+

grafana_api/alerting.py

@@ -965,9 +965,7 @@ def test_backtest_rule(self, condition: str, data_queries: list) -> dict:
             else:
                 return api_call
         else:
-            logging.error(
-                "There is no condition or data_queries defined."
-            )
+            logging.error("There is no condition or data_queries defined.")
             raise ValueError
 
     def delete_ngalert_organization_configuration(self):

grafana_api/datasource.py

@@ -1059,3 +1059,82 @@ def update_datasource_cache(
                 "There is no uid or the right datasource_cache object defined."
             )
             raise ValueError
+
+
+class DatasourceLabelBasedAccessControl:
+    """The class includes all necessary methods to access the Grafana datasource label based access control for teams API endpoints. It's required that the API token got the corresponding datasource access rights. Please check the used methods docstring for the necessary access rights. The functionality is a Grafana Cloud feature. Only cloud Loki data sources are supported
+
+    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 get_lbac_rules_for_datasource(self, uid: str) -> list:
+        """The method includes a functionality to get all datasource label based access control rules for team specified by the datasource uid
+
+        Args:
+            uid (str): Specify the uid of the datasource
+
+        Required Permissions:
+            Action: datasources:read
+            Scope: [datasources:*, datasources:uid:*, datasources:uid:<id>]
+
+        Raises:
+            ValueError: Missed specifying a necessary value
+            Exception: Unspecified error by executing the API call
+
+        Returns:
+            api_call (list): Returns all LBAC rules
+        """
+
+        if len(uid) != 0:
+            api_call: list = Api(self.grafana_api_model).call_the_api(
+                f"{APIEndpoints.DATASOURCES.value}/{uid}/lbac/teams",
+            )
+
+            if api_call is None:
+                logging.error(f"Check the error: {api_call}.")
+                raise Exception
+            else:
+                return api_call
+        else:
+            logging.error("There is no uid defined.")
+            raise ValueError
+
+    def update_lbac_rules_for_datasource(self, uid: str) -> dict:
+        """The method includes a functionality to enable the datasource cache specified by the datasource uid
+
+        Args:
+            uid (str): Specify the uid of the datasource
+
+        Required Permissions:
+            Action: datasources:write, datasources.permissions:write
+            Scope: [datasources:*, datasources:uid:*, datasources:uid:<id>]
+
+        Raises:
+            ValueError: Missed specifying a necessary value
+            Exception: Unspecified error by executing the API call
+
+        Returns:
+            api_call (dict): Returns a datasource
+        """
+
+        if len(uid) != 0:
+            api_call: dict = Api(self.grafana_api_model).call_the_api(
+                f"{APIEndpoints.DATASOURCES.value}/{uid}/lbac/teams",
+                RequestsMethods.POST,
+            )
+
+            if api_call == dict() or api_call.get("dataSourceID") is None:
+                logging.error(f"Check the error: {api_call}.")
+                raise Exception
+            else:
+                return api_call
+        else:
+            logging.error("There is no uid defined.")
+            raise ValueError

grafana_api/user.py

@@ -71,7 +71,7 @@ def search_users_with_paging(
         results_per_page: int = 1000,
         page: int = 1,
         query: str = None,
-        sort: str = None
+        sort: str = None,
     ) -> dict:
         """The method includes a functionality to get all Grafana system users specified by the optional results_per_page, page, query, sort and general paging functionality
 

setup.py

@@ -8,7 +8,7 @@
 
 setuptools.setup(
     name="grafana-api-sdk",
-    version="0.7.2",
+    version="0.8.0",
     author="Pascal Zimmermann",
     author_email="[email protected]",
     description="A Grafana API SDK",

tests/integrationtest/test_service_account.py

@@ -65,8 +65,15 @@ def test_lifecycle_service_account(self):
         self.assertEqual(
             2, len(self.service_account.search_service_account().get("serviceAccounts"))
         )
-        self.service_account.create_service_account_token_by_id(service_account.get("id"), "Test", "Viewer")
-        self.assertEqual(1, self.service_account.search_service_account().get("serviceAccounts")[0].get("tokens"))
+        self.service_account.create_service_account_token_by_id(
+            service_account.get("id"), "Test", "Viewer"
+        )
+        self.assertEqual(
+            1,
+            self.service_account.search_service_account()
+            .get("serviceAccounts")[0]
+            .get("tokens"),
+        )
 
         self.service_account.delete_service_account(service_account.get("id"))
         self.assertEqual(

tests/unittests/test_datasource.py

@@ -12,6 +12,7 @@
     DatasourcePermissions,
     DatasourceLegacyPermissions,
     DatasourceQueryResourceCaching,
+    DatasourceLabelBasedAccessControl,
 )
 
 
@@ -1022,3 +1023,79 @@ def test_update_datasource_cache_no_valid_result(self, call_the_api_mock):
 
         with self.assertRaises(Exception):
             datasource.update_datasource_cache("test", datasource_cache)
+
+
+class DatasourceLabelBasedAccessControlTestCase(TestCase):
+    @patch("grafana_api.api.Api.call_the_api")
+    def test_get_lbac_rules_for_datasource(self, call_the_api_mock):
+        model: APIModel = APIModel(host=MagicMock(), token=MagicMock())
+        datasource: DatasourceLabelBasedAccessControl = (
+            DatasourceLabelBasedAccessControl(grafana_api_model=model)
+        )
+
+        call_the_api_mock.return_value = list([{"id": 1}])
+
+        self.assertEqual([{"id": 1}], datasource.get_lbac_rules_for_datasource("test"))
+
+    @patch("grafana_api.api.Api.call_the_api")
+    def test_get_lbac_rules_for_datasource_no_uid(self, call_the_api_mock):
+        model: APIModel = APIModel(host=MagicMock(), token=MagicMock())
+        datasource: DatasourceLabelBasedAccessControl = (
+            DatasourceLabelBasedAccessControl(grafana_api_model=model)
+        )
+
+        call_the_api_mock.return_value = None
+
+        with self.assertRaises(ValueError):
+            datasource.get_lbac_rules_for_datasource("")
+
+    @patch("grafana_api.api.Api.call_the_api")
+    def test_get_lbac_rules_for_datasource_no_valid_rules(self, call_the_api_mock):
+        model: APIModel = APIModel(host=MagicMock(), token=MagicMock())
+        datasource: DatasourceLabelBasedAccessControl = (
+            DatasourceLabelBasedAccessControl(grafana_api_model=model)
+        )
+
+        call_the_api_mock.return_value = None
+
+        with self.assertRaises(Exception):
+            datasource.get_lbac_rules_for_datasource("test")
+
+    @patch("grafana_api.api.Api.call_the_api")
+    def test_update_lbac_rules_for_datasource(self, call_the_api_mock):
+        model: APIModel = APIModel(host=MagicMock(), token=MagicMock())
+        datasource: DatasourceLabelBasedAccessControl = (
+            DatasourceLabelBasedAccessControl(grafana_api_model=model)
+        )
+
+        call_the_api_mock.return_value = dict({"dataSourceID": 1})
+
+        self.assertEqual(
+            {"dataSourceID": 1}, datasource.update_lbac_rules_for_datasource("test")
+        )
+
+    @patch("grafana_api.api.Api.call_the_api")
+    def test_update_lbac_rules_for_datasource_no_uid(self, call_the_api_mock):
+        model: APIModel = APIModel(host=MagicMock(), token=MagicMock())
+        datasource: DatasourceLabelBasedAccessControl = (
+            DatasourceLabelBasedAccessControl(grafana_api_model=model)
+        )
+
+        call_the_api_mock.return_value = None
+
+        with self.assertRaises(ValueError):
+            datasource.update_lbac_rules_for_datasource("")
+
+    @patch("grafana_api.api.Api.call_the_api")
+    def test_update_lbac_rules_for_datasource_no_update_possible(
+        self, call_the_api_mock
+    ):
+        model: APIModel = APIModel(host=MagicMock(), token=MagicMock())
+        datasource: DatasourceLabelBasedAccessControl = (
+            DatasourceLabelBasedAccessControl(grafana_api_model=model)
+        )
+
+        call_the_api_mock.return_value = dict()
+
+        with self.assertRaises(Exception):
+            datasource.update_lbac_rules_for_datasource("test")