-
Notifications
You must be signed in to change notification settings - Fork 2
/
client.py
58 lines (46 loc) · 2.18 KB
/
client.py
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
""" Used for communication with Camunda Engine """
import requests
import config
class CamundaClient:
""" Class used for communication with camunda engine """
def __init__(self, url=config.CAMUNDA_ENGINE_URI):
self.url = url
# Status
@classmethod
def status_code_successful(cls, status_code: int) -> bool:
"""Assert whether an HTTP status code is successful
:param status_code: HTTP status code
:return: True or False
"""
return str(status_code)[0] == '2'
### Section: Deployment & Starting ##
# Deploying a single process
def deploy_process(self, path_bpmn_file: str) -> str:
"""Send a process to Camunda Engine and retrieve process id in Camunda Engine.
:param path_bpmn_file: path of bpmn file on filesystem of server
:return: id of process in camunda engine
"""
with open(path_bpmn_file, 'r', encoding="UTF-8") as bpmn_file:
multipart_form_data = {
'deployment-name': (None, 'store'),
'data': (path_bpmn_file.split("/")[len(path_bpmn_file.split("/")) - 1], bpmn_file),
}
response = requests.post(self.url + "/deployment/create", files=multipart_form_data, timeout=5)
assert self.status_code_successful(response.status_code)
assert len(response.json().get('deployedProcessDefinitions')) == 1
new_process_id = None
for elem in response.json().get('deployedProcessDefinitions'):
new_process_id = elem
return new_process_id
# Starting a single instance
def start_instance(self, process_id: str) -> str:
"""Start an instance with a certain camunda process id.
:param process_id: specify which process with !camunda process id!
:return: instance id in camunda engine
"""
headers = {'Content-Type': 'application/json'}
response = requests.post(self.url + "/process-definition/" + str(process_id) + "/start",
headers=headers,
timeout=5)
assert self.status_code_successful(response.status_code)
return response.json().get('id')