First python utility script to get all the users from Keycloak server

python/get-all-users.py

@@ -0,0 +1,72 @@
+#/bin/python3
+import requests
+import threading
+import csv
+
+# To be completed with access data
+username = "xxxx"
+password = "xxxxxxxx"
+# Endpoint data
+server = "https://accounts.pre.d4science.org"
+#server = "http://localhost:8080"
+realm = "d4science"
+
+# Fine tuning in case of timeout
+bunchsize = 500
+
+tokenurl = server + "/auth/realms/master/protocol/openid-connect/token"
+baseurl = server + "/auth/admin/realms/" + realm
+
+payload='grant_type=password&client_id=admin-cli&username=' + username + '&password=' + password
+headers = {
+  'Content-Type': 'application/x-www-form-urlencoded'
+}
+
+accesstoken = requests.request("POST", tokenurl, headers=headers, data=payload).json()['access_token']
+
+headers = {
+  'Authorization': 'Bearer ' + accesstoken
+}
+
+def thread_function(index, first):
+    end = first + bunchsize - 1 
+    print("Starting thread for users window: " + str(first) + " - " + str(end))
+    response = requests.request("GET", baseurl + "/users?briefRepresentation=true&first=" + str(first) + "&max=" + str(bunchsize), headers=headers)
+    if response.status_code is 200:
+        userwindows[index] = response.json()
+    else:
+        print("Error for user window thread " + str(first) + " - " + str(end) + ": " + response.text)
+        userwindows[index] = None
+
+userscount = requests.request("GET", baseurl + "/users/count", headers=headers).json()
+
+print("Users count: " + str(userscount))
+
+userwindows = [None] * (int(userscount / bunchsize) + 1)
+index = 0
+threads = list()
+for first in range(0, userscount, bunchsize):
+    x = threading.Thread(target=thread_function, args=(index, first))
+    threads.append(x)
+    x.start()
+    index += 1
+
+for thread in threads:
+    thread.join()
+
+print("Download complete, start CSV export to file...")
+
+witherror = False
+with open('users.csv', 'w', newline='') as file:
+    writer = csv.writer(file)
+    for window in userwindows:
+        if window is not None:
+            for user in window:
+                writer.writerow([user['username'], user.get('email','<not-found>')])
+        else:
+            witherror = True
+
+if not witherror:
+    print("... Export completed successfully")
+else:
+    print("... Export completed with errors")