forked from QonicOpen/ApiPythonSample
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsample.py
More file actions
172 lines (149 loc) · 5.66 KB
/
Copy pathsample.py
File metadata and controls
172 lines (149 loc) · 5.66 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
from random import randint
import requests
import re
import uuid
from oauth import login
apiUrl = "https://api.qonic.com/v1/"
tokenResponse = login(
issuer="https://release-qonic.eu.auth0.com",
client_id="9Jtp6GGNqPPJzvqNKRoQJ66A9juVbE8A",
redirect_uri="http://localhost:34362",
scope="openid profile email",
audience="https://api.qonic.com")
class ModificationInputError:
def __init__(self, guid, field, error, description):
self.guid = guid
self.field = field
self.error = error
self.description = description
def __str__(self):
return f"{self.guid}: {self.field}: {self.error}: {self.description}"
def __repr__(self):
return f"{self.guid}: {self.field}: {self.error}: {self.description}"
class ApiError:
def __init__(self, Error, ErrorDetails):
self.error = Error
self.details = ErrorDetails
def __str__(self):
return f"{self.error}: {self.details}"
def __repr__(self):
return f"{self.error}: {self.details}"
def handleErrorResponse(response: requests.Response):
try:
apiError = ApiError(**response.json())
print(apiError)
except Exception as err:
print(f"Error occurred while processing error response: {err}")
def sendGetRequest(path, params=None):
try:
response = requests.get(f"{apiUrl}{path}", params=params, headers={"Authorization": f"Bearer {tokenResponse.access_token}"})
response.raise_for_status()
except requests.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
handleErrorResponse(response)
exit()
except Exception as err:
print(f"Other error occurred: {err}")
exit()
return response.json()
def sendPostRequest(path, data=None, json=None, params=None, sessionId=str):
try:
response = requests.post(f"{apiUrl}{path}", data=data, json=json, params=params, headers={"Authorization": f"Bearer {tokenResponse.access_token}", "X-Client-Session-Id": sessionId})
response.raise_for_status()
return response
except requests.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
handleErrorResponse(response)
exit()
except Exception as err:
print(f"Other error occurred: {err}")
exit()
projectsJson = sendGetRequest("projects")
print("your projects:")
for project in projectsJson["projects"]:
print(f"{project['id']} - {project['name']}")
print()
projectId = input("Enter a project id: ")
print()
modelsJson = sendGetRequest(f"projects/{projectId}/models")
print("your models:")
for model in modelsJson["models"]:
print(f"{model['id']} - {model['name']}")
print()
modelId = input("Enter a model id: ")
print()
availableDataJson = sendGetRequest(f"projects/{projectId}/models/{modelId}/external-query-available-data")
print("fields:")
for field in availableDataJson["fields"]:
print(f"{field}")
print()
print("Quering the Guid, Class, Name and FireRating fields, filtered on Class Beam...")
print()
# Query the Guid, Class, Name and FireRating fields
fieldsStr = "Guid Class Name FireRating"
params = list(map(lambda field: ("fields", field), re.split(r'[;,\s]+', fieldsStr)))
# Filter on class Beam
params.append(("filters[Class]", "Beam"))
propertiesJson = sendGetRequest(f"projects/{projectId}/models/{modelId}/external-query", params)
print()
for row in propertiesJson["result"]:
print(f"{row}")
print()
print("Starting modification session")
sessionId = str(uuid.uuid4())
sendPostRequest(f"projects/{projectId}/models/{modelId}/start-session", sessionId=sessionId)
try:
fireRating = f"F{randint(1, 200)}"
print(f"Modifying FireRating of first row to {fireRating}")
changes = {
"values": {
"FireRating": {
propertiesJson["result"][0]["Guid"]: {
"PropertySet": "Pset_BeamCommon",
"Value": fireRating
}
}
}
}
response = sendPostRequest(f"projects/{projectId}/models/{modelId}/external-data-modification", json=changes, sessionId=sessionId)
errors = list(map(lambda json: ModificationInputError(**json), response.json()["errors"]))
if len(errors) > 0:
print(str(errors))
exit()
finally:
print("Closing modification session")
sendPostRequest(f"projects/{projectId}/models/{modelId}/end-session", sessionId=sessionId)
print("Modification is done")
print()
print("Quering data again")
propertiesJson = sendGetRequest(f"projects/{projectId}/models/{modelId}/external-query", params)
print("Showing only the first row:")
print(propertiesJson["result"][0])
print()
print("Starting modification session to reset value")
sessionId = str(uuid.uuid4())
sendPostRequest(f"projects/{projectId}/models/{modelId}/start-session", sessionId=sessionId)
try:
fireRating = f"F{randint(1, 200)}"
print(f"Clearing FireRating of first row")
changes = {
"values": {
"FireRating": {
propertiesJson["result"][0]["Guid"]: None
}
}
}
response = sendPostRequest(f"projects/{projectId}/models/{modelId}/external-data-modification", json=changes, sessionId=sessionId)
errors = list(map(lambda json: ModificationInputError(**json), response.json()["errors"]))
if len(errors) > 0:
print(errors)
exit()
finally:
print("Closing modification session")
sendPostRequest(f"projects/{projectId}/models/{modelId}/end-session", sessionId=sessionId)
print("Modification is done")
print()
print("Quering data again")
propertiesJson = sendGetRequest(f"projects/{projectId}/models/{modelId}/external-query", params)
print("Showing only the first row:")
print(propertiesJson["result"][0])