diff --git a/docs/content/grafana_api/datasource.md b/docs/content/grafana_api/datasource.md index 73d6a6a..01a96fe 100644 --- a/docs/content/grafana_api/datasource.md +++ b/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) @@ -852,3 +855,80 @@ The method includes a functionality to update the datasource cache specified by - `api_call` _dict_ - Returns a datasource + + +## 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 + + + +#### 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:] + + +**Raises**: + +- `ValueError` - Missed specifying a necessary value +- `Exception` - Unspecified error by executing the API call + + +**Returns**: + +- `api_call` _list_ - Returns all LBAC rules + + + +#### 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:] + + +**Raises**: + +- `ValueError` - Missed specifying a necessary value +- `Exception` - Unspecified error by executing the API call + + +**Returns**: + +- `api_call` _dict_ - Returns a datasource + diff --git a/grafana_api/alerting.py b/grafana_api/alerting.py index b499508..849fdf3 100644 --- a/grafana_api/alerting.py +++ b/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): diff --git a/grafana_api/datasource.py b/grafana_api/datasource.py index c117c8d..3492e1f 100644 --- a/grafana_api/datasource.py +++ b/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:] + + 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:] + + 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 diff --git a/grafana_api/user.py b/grafana_api/user.py index 48576c8..471eea0 100644 --- a/grafana_api/user.py +++ b/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 diff --git a/setup.py b/setup.py index 58294c3..cf91b42 100644 --- a/setup.py +++ b/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="info@theiotstudio.com", description="A Grafana API SDK", diff --git a/tests/integrationtest/test_service_account.py b/tests/integrationtest/test_service_account.py index 2a699bc..566ead0 100644 --- a/tests/integrationtest/test_service_account.py +++ b/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( diff --git a/tests/unittests/test_datasource.py b/tests/unittests/test_datasource.py index 69d1b2b..51c0664 100644 --- a/tests/unittests/test_datasource.py +++ b/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")