Skip to content

Commit ff7f682

Browse files
committed
added execution server sample
1 parent c19b56a commit ff7f682

1 file changed

Lines changed: 55 additions & 0 deletions

File tree

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
import os
2+
import requests
3+
import json
4+
5+
6+
class QualiApiError(Exception):
7+
pass
8+
9+
10+
class QualiAPISession:
11+
def __init__(self, host, username='', password='', domain='Global', token_id='', port=9000):
12+
self._base_url = "http://{}:{}".format(host, port)
13+
self.session = requests.Session()
14+
self.login_set_auth_header(domain, password, token_id, username)
15+
16+
def login_set_auth_header(self, domain, password, token_id, username):
17+
if token_id:
18+
headers = {"token": token_id, "domain": domain}
19+
elif username and password:
20+
headers = {"username": username, "password": password, "domain": domain}
21+
else:
22+
raise ValueError("Must supply Username / Password OR token_id")
23+
login_result = requests.put(f"{self._base_url}/API/Auth/Login", headers)
24+
if not login_result.ok:
25+
raise QualiApiError(f"Failed API Login. Status: {login_result.status_code}. Reason: {login_result.text}")
26+
token_str = login_result.text[1:-1]
27+
auth_header = {"Authorization": f"Basic {token_str}"}
28+
self.session.headers.update(auth_header)
29+
30+
@staticmethod
31+
def _validate_response(response: requests.Response):
32+
if not response.ok:
33+
raise QualiApiError(f"Failed api call. Status: {response.status_code}. Reason: {response.text}")
34+
35+
def get_execution_servers(self):
36+
url = f"{self._base_url}/API/Manage/ExecutionServers"
37+
response = self.session.get(url)
38+
self._validate_response(response)
39+
return response.json()
40+
41+
def get_execution_server_details(self, execution_server_name):
42+
url = f"{self._base_url}/API/Manage/ExecutionServers/{execution_server_name}"
43+
response = self.session.get(url)
44+
self._validate_response(response)
45+
return response.json()
46+
47+
def get_all_server_details(self):
48+
all_servers = self.get_execution_servers()
49+
return [self.get_execution_server_details(x["Name"]) for x in all_servers]
50+
51+
52+
if __name__ == "__main__":
53+
api = QualiAPISession("localhost", "admin", "admin", "Global")
54+
all_server_details = api.get_all_server_details()
55+
print(json.dumps(all_server_details, indent=4))

0 commit comments

Comments
 (0)