Create workflows.py
taylor_socfortress committed
Jul 10, 2023 at 16:44 UTC
7475ca0322bca68a82644a1dddb443e97ea48469
1 file changed
+178
backend/app/services/Shuffle/workflows.py
new
+178
@@ -0,0 +1,178 @@
1
+from typing import Dict
2
+import requests
3
+from loguru import logger
4
+from app.services.Shuffle.universal import UniversalService
5
+
6
+
7
+class WorkflowsService:
8
+ """
9
+ A service class that encapsulates the logic for pulling workflows from Shuffle.
10
+ """
11
+
12
+ def __init__(self):
13
+ self._collect_shuffle_details()
14
+ self.session = requests.Session()
15
+ self.session.headers.update({"Authorization" : f"Bearer {self.connector_api_key}"})
16
+
17
+ def _collect_shuffle_details(self):
18
+ self.connector_url, self.connector_api_key = UniversalService().collect_shuffle_details("Shuffle")
19
+
20
+ def _are_details_collected(self) -> bool:
21
+ return all([self.connector_url, self.connector_api_key])
22
+
23
+ def _send_request(self, url: str):
24
+ return self.session.get(
25
+ url,
26
+ verify=False,
27
+ )
28
+
29
+ def collect_workflows(self) -> Dict[str, object]:
30
+ """
31
+ Collects the workflows from Shuffle.
32
+
33
+ Returns:
34
+ dict: A dictionary containing the success status, a message and potentially the workflows.
35
+ """
36
+ if not self._are_details_collected():
37
+ return {
38
+ "message": "Failed to collect Shuffle details",
39
+ "success": False,
40
+ }
41
+
42
+ workflows = self._collect_workflows()
43
+ if not workflows["success"]:
44
+ return workflows
45
+
46
+ return {
47
+ "message": "Successfully collected workflows",
48
+ "success": True,
49
+ "workflows": workflows["workflows"],
50
+ }
51
+
52
+ def _handle_request_error(self, err):
53
+ logger.error(f"Failed to collect workflows from Shuffle: {err}")
54
+ return {
55
+ "message": "Failed to collect workflows from Shuffle",
56
+ "success": False,
57
+ }
58
+
59
+ def _collect_workflows(self) -> Dict[str, object]:
60
+ """
61
+ Collects the workflows from Shuffle.
62
+
63
+ Returns:
64
+ dict: A dictionary containing the success status, a message and potentially the workflows.
65
+ """
66
+ try:
67
+ response = self._send_request(f"{self.connector_url}/api/v1/workflows")
68
+ response.raise_for_status()
69
+ except requests.exceptions.HTTPError as err:
70
+ return self._handle_request_error(err)
71
+
72
+ return {
73
+ "message": "Successfully collected workflows from Shuffle",
74
+ "success": True,
75
+ "workflows": response.json(),
76
+ }
77
+
78
+ def collect_workflow_details(self) -> Dict[str, object]:
79
+ """
80
+ Collects the workflow ID and workflow name from Shuffle.
81
+
82
+ Returns:
83
+ dict: A dictionary containing the success status, a message and potentially the workflow IDs.
84
+ """
85
+ if not self._are_details_collected():
86
+ return {
87
+ "message": "Failed to collect Shuffle details",
88
+ "success": False,
89
+ }
90
+
91
+ workflows = self._collect_workflow_details()
92
+ if not workflows["success"]:
93
+ return workflows
94
+
95
+ return {
96
+ "message": "Successfully collected workflow details",
97
+ "success": True,
98
+ "workflows": workflows["workflows"],
99
+ }
100
+
101
+ def _collect_workflow_details(self) -> Dict[str, object]:
102
+ """
103
+ Collects the workflow ID and workflow name from Shuffle.
104
+
105
+ Returns:
106
+ dict: A dictionary containing the success status, a message and potentially the workflow IDs.
107
+ """
108
+ try:
109
+ response = self._send_request(f"{self.connector_url}/api/v1/workflows")
110
+ response.raise_for_status()
111
+ except requests.exceptions.HTTPError as err:
112
+ return self._handle_request_error(err)
113
+
114
+ workflows = response.json()
115
+ workflow_details = []
116
+ for workflow in workflows:
117
+ workflow_details.append({
118
+ "workflow_id": workflow["id"],
119
+ "workflow_name": workflow["name"]
120
+ })
121
+
122
+ return {
123
+ "message": "Successfully collected workflow details from Shuffle",
124
+ "success": True,
125
+ "workflows": workflow_details,
126
+ }
127
+
128
+ def collect_workflow_executions_status(self, workflow_id: str) -> Dict[str, object]:
129
+ """
130
+ Collects the execution status of a Shuffle Workflow by its ID.
131
+
132
+ Returns:
133
+ dict: A dictionary containing the success status, a message and potentially the workflow execution status.
134
+ """
135
+ if not self._are_details_collected():
136
+ return {
137
+ "message": "Failed to collect Shuffle details",
138
+ "success": False,
139
+ }
140
+
141
+ executions = self._collect_workflow_executions_status(workflow_id)
142
+ if not executions["success"]:
143
+ return executions
144
+
145
+ return {
146
+ "message": "Successfully collected workflow executions",
147
+ "success": True,
148
+ "executions": executions["executions"],
149
+ }
150
+
151
+ def _collect_workflow_executions_status(self, workflow_id: str) -> Dict[str, object]:
152
+ """
153
+ Collects the execution status of a Shuffle Workflow by its ID.
154
+
155
+ Returns:
156
+ dict: A dictionary containing the success status, a message and potentially the workflow execution status.
157
+ """
158
+ try:
159
+ response = self._send_request(f"{self.connector_url}/api/v1/workflows/{workflow_id}/executions")
160
+ response.raise_for_status()
161
+ except requests.exceptions.HTTPError as err:
162
+ return self._handle_request_error(err)
163
+
164
+ executions = response.json()
165
+ if executions:
166
+ status = executions[0]["status"]
167
+ if status is None:
168
+ status = "Never Ran"
169
+ else:
170
+ logger.info(f"No Workflow Executions found from {self.connector_url}")
171
+ status = None
172
+
173
+ return {
174
+ "message": "Successfully collected workflow executions from Shuffle",
175
+ "success": True,
176
+ "executions": status,
177
+
178
+ }