Add support for service accounts

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## unreleased
 
+* Add support for service accounts. Thanks, @einar-lanfranco and @feerbau.
+
 
 ## 3.10.0 (2023-10-30)
 

examples/service-account.py

@@ -0,0 +1,71 @@
+"""
+About
+=====
+
+Example program for interacting with the Grafana Service Account.
+
+Usage
+=====
+
+Make sure to adjust `GrafanaApi` options at the bottom of the
+file.
+And replace aaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa with the uid of the folder where you want to add the service account.
+
+Synopsis
+========
+::
+
+    source .venv/bin/activate
+    python examples/service-account.py
+
+"""
+import uuid
+
+from grafana_client import GrafanaApi
+from grafana_client.client import GrafanaClientError
+
+
+def run_conversation(grafana: GrafanaApi):
+    # print("Grafana address")
+    # print(grafana.client.url)
+
+    # print("Health check")
+    # print(grafana.health.check())
+
+    print("Create Service Account")
+    sa = {"name": "Sytem Account DEMO", "role": "Viewer"}
+
+    # Add a new service account
+    try:
+        response = grafana.serviceaccount.create(sa)
+        print(response)
+    except GrafanaClientError as ex:
+        print(f"ERROR: {ex}")
+
+    # Add a new token to service account, you need to writedown the token['key'] to use it later
+    # Genera an UUID
+    uuid_to_token = uuid.uuid4()
+    token_name = "%s" % (uuid_to_token)
+
+    try:
+        token = grafana.serviceaccount.create_token(response["id"], {"name": token_name})
+        print(token)
+    except GrafanaClientError as ex:
+        print(f"ERROR: {ex}")
+
+    # Add Viewer permissions to service account created to folder with uid aaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa
+    uid_folder = "aaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
+    permissions = {"permission": "View"}
+    sa_id = response["id"]
+    grafana.folder.update_folder_permissions_for_user(uid_folder, sa_id, permissions)
+
+    # Search service account by name
+    search_sa = grafana.serviceaccount.search_one(service_account_name="Sytem Account DEMO")
+    print(search_sa)
+
+
+if __name__ == "__main__":
+    # Connect to custom Grafana instance.
+    grafana = GrafanaApi.from_url(url="localhost:3000", credential=("admin", "admin"))
+
+    run_conversation(grafana)

grafana_client/api.py

@@ -26,6 +26,7 @@
     Plugin,
     Rbac,
     Search,
+    ServiceAccount,
     Snapshots,
     Teams,
     User,
@@ -80,6 +81,7 @@ def __init__(
         self.snapshots = Snapshots(self.client)
         self.notifications = Notifications(self.client)
         self.plugin = Plugin(self.client)
+        self.serviceaccount = ServiceAccount(self.client)
 
     def connect(self):
         try:

grafana_client/elements/__init__.py

@@ -13,6 +13,31 @@
 from .plugin import Plugin
 from .rbac import Rbac
 from .search import Search
+from .service_account import ServiceAccount
 from .snapshots import Snapshots
 from .team import Teams
 from .user import User, Users
+
+__all__ = (
+    "Admin",
+    "Alerting",
+    "AlertingProvisioning",
+    "Annotations",
+    "Base",
+    "Dashboard",
+    "DashboardVersions",
+    "Datasource",
+    "Folder",
+    "Health",
+    "Notifications",
+    "Organization",
+    "Organizations",
+    "Plugin",
+    "Rbac",
+    "Search",
+    "ServiceAccount",
+    "Snapshots",
+    "Teams",
+    "User",
+    "Users",
+)

grafana_client/elements/folder.py

@@ -100,3 +100,17 @@ def update_folder_permissions(self, uid, items):
         update_folder_permissions_path = "/folders/%s/permissions" % uid
         r = self.client.POST(update_folder_permissions_path, json=items)
         return r
+
+    def update_folder_permissions_for_user(self, uid, user_id, items):
+        """
+
+        :param uid:
+        :param user_id:
+        :param items:
+            {"permission": "View"} or {"permission": "Edit"} or {"permission": ""}
+        :return:
+        """
+
+        update_folder_permissions_path_for_user = "/access-control/folders/%s/users/%s" % (uid, user_id)
+        r = self.client.POST(update_folder_permissions_path_for_user, json=items)
+        return r

grafana_client/elements/service_account.py

@@ -0,0 +1,145 @@
+"""
+https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/
+https://grafana.com/docs/grafana/latest/administration/service-accounts/
+https://grafana.com/docs/grafana/latest/developers/http_api/create-api-tokens-for-org/
+"""
+from .base import Base
+
+
+class ServiceAccount(Base):
+    def __init__(self, client):
+        super(ServiceAccount, self).__init__(client)
+        self.client = client
+        self.path = "/serviceaccounts"
+
+    def get(self, service_account_id):
+        """
+        Get service account by id.
+        https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#get-a-service-account-by-id
+
+        :param service_account_id:
+        :return:
+        """
+        get_actual_user_path = "/serviceaccounts/%s?accesscontrol=true" % (service_account_id)
+        r = self.client.GET(get_actual_user_path)
+        return r
+
+    def create(self, service_account):
+        """
+        Create service account.
+        https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#create-service-account
+
+        :param service_account: {"name": "string", "role": "string"}
+        :return:
+        """
+        create_service_account_path = "/serviceaccounts/"
+        r = self.client.POST(create_service_account_path, json=service_account)
+        return r
+
+    def delete(self, service_account_id):
+        """
+        Delete service account.
+        https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#delete-service-account
+
+        :param service_account:
+        :return:
+        """
+
+        delete_service_account_path = "/serviceaccounts/%s" % (service_account_id)
+        r = self.client.DELETE(delete_service_account_path)
+        return r
+
+    def get_tokens(self, service_account_id):
+        """
+        Get service account tokens.
+        https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#get-service-account-tokens
+
+        :param service_account_id:
+        :return:
+        """
+        service_account_tokens_path = "/serviceaccounts/%s/tokens" % (service_account_id)
+        r = self.client.GET(service_account_tokens_path)
+        return r
+
+    def create_token(self, service_account_id, content):
+        """
+        Create service account tokens
+        https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#create-service-account-tokens
+
+        :param service_account_id:
+        :param service_account_name:
+        :return:
+        """
+        create_service_account_token_path = "/serviceaccounts/%s/tokens" % (service_account_id)
+        r = self.client.POST(create_service_account_token_path, json=content)
+        return r
+
+    def delete_token(self, service_account_id, service_account_token_id):
+        """
+        Delete service account tokens.
+        https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#delete-service-account-tokens
+
+        :param service_account_id:
+        :param service_account_token_id:
+        :return:
+        """
+        delete_service_account_token_path = "/serviceaccounts/%s/tokens/%s" % (
+            service_account_id,
+            service_account_token_id,
+        )
+        r = self.client.DELETE(delete_service_account_token_path)
+        return r
+
+    def search(self, query=None, page=None, perpage=None):
+        """
+
+        :return:
+        """
+        list_of_sa = []
+        sa_on_page = None
+        show_sa_path = "/serviceaccounts/search"
+        params = []
+        if query:
+            params.append("query=%s" % query)
+
+        if page:
+            iterate = False
+            params.append("page=%s" % page)
+        else:
+            iterate = True
+            params.append("page=%s")
+            page = 1
+
+        if perpage:
+            params.append("perpage=%s" % perpage)
+
+        show_sa_path += "?"
+        show_sa_path += "&".join(params)
+        if iterate:
+            while True:
+                url = show_sa_path % page
+                sa_on_page = self.client.GET(url)
+                list_of_sa.append(sa_on_page)
+                if not sa_on_page["serviceAccounts"]:
+                    break
+                page += 1
+        else:
+            sa_on_page = self.client.GET(show_sa_path)
+            list_of_sa.append(sa_on_page)
+
+        return list_of_sa
+
+    def search_one(self, service_account_name=""):
+        """
+        Find a single service account by name. Raises errors on multiple or no matches.
+
+        :param service_account_name:
+        :return:
+        """
+        s = self.search(query=service_account_name)[0]
+        if s["totalCount"] == 1:
+            return s["serviceAccounts"][0]
+        elif s["totalCount"] > 1:
+            raise ValueError("More than one service account matched")
+        else:
+            raise ValueError("No service account matched")

test/elements/__init__.py

@@ -0,0 +1 @@
+

test/elements/test_folder.py

@@ -246,6 +246,22 @@ def test_update_folder_permissions(self, m):
         )
         self.assertEqual(folder["message"], "Folder permissions updated")
 
+    @requests_mock.Mocker()
+    def test_update_folder_permissions_for_user(self, m):
+        m.post(
+            "http://localhost/api/access-control/folders/nErXDvCkzz/users/12345",
+            json={"message": "Folder permissions updated"},
+        )
+        folder = self.grafana.folder.update_folder_permissions_for_user(
+            uid="nErXDvCkzz",
+            user_id="12345",
+            items=[
+                {"permission": "View"},
+                {"permission": "Edit"},
+            ],
+        )
+        self.assertEqual(folder["message"], "Folder permissions updated")
+
     @requests_mock.Mocker()
     def test_delete_folder(self, m):
         m.delete("http://localhost/api/folders/nErXDvCkzz", json={"message": "Folder deleted"})