Terarea  2
The automation project
Loading...
Searching...
No Matches
applets.py
Go to the documentation of this file.
1"""
2The file that contains every endpoints for applets
3"""
4
5from typing import Union, List, Dict, Any
6import json
7from fastapi import Response, Request
8from display_tty import Disp, TOML_CONF, FILE_DESCRIPTOR, SAVE_TO_FILE, FILE_NAME
9from .. import constants as CONST
10from ..runtime_data import RuntimeData
11from ..http_codes import HCI
12
13class Applets:
14 """
15 The class that contains every methods for applets
16 """
17 def __init__(self, runtime_data: RuntimeData, success: int = 0, error: int = 84, debug: bool = False) -> None:
18 """
19 The constructor of the Applets class
20 """
21 self.debug: bool = debug
22 self.success: int = success
23 self.error: int = error
24 self.runtime_data_initialised: RuntimeData = runtime_data
25 self.disp: Disp = Disp(
26 TOML_CONF,
27 SAVE_TO_FILE,
28 FILE_NAME,
29 FILE_DESCRIPTOR,
30 debug=self.debug,
31 logger=self.__class__.__name__
32 )
33
34 # Endpoints for my applets page
35
36 async def get_applet_by_id(self, request: Request, applet_id: str) -> Response:
37 """
38 Get applet by id
39 """
40 title = "Get applet by id"
41 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
42 request
43 )
44 if not token:
45 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
46 title
47 )
48 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
49 token
50 ) is False:
51 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
52 title
53 )
54 self.disp.log_debug(f"Token = {token}", title)
55 user_id: Union[str, Response] = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
56 title=title,
57 token=token
58 )
59 if isinstance(user_id, Response):
60 return user_id
61 applet_data = self.runtime_data_initialised.database_link.get_data_from_table(
62 CONST.TAB_ACTIONS,
63 "*",
64 [f"id='{applet_id}'", f"user_id='{user_id}'"]
65 )
66 if not applet_data or isinstance(applet_data, int):
67 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
68 title=title,
69 message="Your applet was not found.",
70 resp="not found",
71 token=token,
72 error=True
73 )
74 return HCI.not_found(
75 content=body,
76 content_type=CONST.CONTENT_TYPE,
77 headers=self.runtime_data_initialised.json_header
78 )
79 applet_data[0]["trigger"] = json.loads(applet_data[0]["trigger"])
80 applet_data[0]["consequences"] = json.loads(applet_data[0]["consequences"])
81 self.disp.log_debug(f"Applet found: {applet_data}", title)
82 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
83 title=title,
84 message=applet_data[0],
85 resp="success",
86 token=token
87 )
88 return HCI.success(
89 content=body,
90 content_type=CONST.CONTENT_TYPE,
91 headers=self.runtime_data_initialised.json_header
92 )
93
94 async def get_my_applets(self, request: Request) -> Response:
95 """
96 Get my applets
97 """
98 title = "Get my applets"
99 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
100 request
101 )
102 if not token:
103 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
104 title
105 )
106 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
107 token
108 ) is False:
109 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
110 title
111 )
112 self.disp.log_debug(f"Token = {token}", title)
113 user_id = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
114 title,
115 token
116 )
117 if isinstance(user_id, Response):
118 return user_id
119 applets_data = self.runtime_data_initialised.database_link.get_data_from_table(
120 CONST.TAB_ACTIONS,
121 "*",
122 f"user_id='{user_id}'"
123 )
124 if not applets_data or isinstance(applets_data, int):
125 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
126 title=title,
127 message="Your applets were not found.",
128 resp="not found",
129 token=token,
130 error=True
131 )
132 return HCI.not_found(
133 content=body,
134 content_type=CONST.CONTENT_TYPE,
135 headers=self.runtime_data_initialised.json_header
136 )
137 for _, applet in enumerate(applets_data):
138 applet["trigger"] = json.loads(applet["trigger"])
139 applet["consequences"] = json.loads(applet["consequences"])
140 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
141 title=title,
142 message=applets_data,
143 resp="success",
144 token=token
145 )
146 return HCI.success(
147 content=body,
148 content_type=CONST.CONTENT_TYPE,
149 headers=self.runtime_data_initialised.json_header
150 )
151
152 async def get_my_applets_by_tags(self, request: Request, tags: str) -> Response:
153 """
154 Get my applets by tags
155 """
156 title = "Get my applets by tags"
157 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
158 request
159 )
160 if not token:
161 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
162 title
163 )
164 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
165 token
166 ) is False:
167 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
168 title
169 )
170 self.disp.log_debug(f"Token = {token}", title)
171 user_id = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
172 title,
173 token
174 )
175 if isinstance(user_id, Response):
176 return user_id
177 if not tags or tags == "":
178 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
179 title,
180 token
181 )
182 tags_list = tags.split("+")
183 applets_data = self.runtime_data_initialised.database_link.get_data_from_table(
184 CONST.TAB_ACTIONS,
185 "*",
186 f"user_id='{user_id}'"
187 )
188 if not applets_data or isinstance(applets_data, int):
189 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
190 title=title,
191 message="Applets not found.",
192 resp="not found",
193 token=token,
194 error=True
195 )
196 return HCI.not_found(
197 content=body,
198 content_type=CONST.CONTENT_TYPE,
199 headers=self.runtime_data_initialised.json_header
200 )
201 self.disp.log_debug(f"Applet found: {applets_data}", title)
202 filtered_applets: list[dict] = []
203 for _, applet in enumerate(applets_data):
204 for _, element in enumerate(tags_list):
205 if element in applet["tags"]:
206 filtered_applets.append(applet)
207 self.disp.log_debug(f"Applet found with tags: {filtered_applets}", title)
208 for _, applet in enumerate(filtered_applets):
209 applet["trigger"] = json.loads(applet["trigger"])
210 applet["consequences"] = json.loads(applet["consequences"])
211 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
212 title=title,
213 message=filtered_applets,
214 resp="success",
215 token=token
216 )
217 return HCI.success(
218 content=body,
219 content_type=CONST.CONTENT_TYPE,
220 headers=self.runtime_data_initialised.json_header
221 )
222
223 # Endpoints for Create/Modify applets page
224
225 async def create_applet(self, request: Request) -> Response:
226 """
227 Create applet
228 """
229 title = "Create applet"
230
231 # Token getter
232 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
233 request
234 )
235 if not token:
236 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
237 title
238 )
239 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
240 token
241 ) is False:
242 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
243 title
244 )
245 self.disp.log_debug(f"Token = {token}", title)
246
247 # Request Body getter
248 request_body = await self.runtime_data_initialised.boilerplate_incoming_initialised.get_body(request)
249 if not request_body or not all(key in request_body for key in ("name", "trigger", "consequences", "tags", "description", "colour")):
250 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
251 if request_body["name"] == "" or request_body["description"] == "" or request_body["colour"] == "" or len(request_body["trigger"]) == 0 or len(request_body["consequences"]) == 0:
252 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
253 self.disp.log_debug(f"Request body: {request_body}", title)
254
255 # User id getter
256 user_id: Union[str, Response] = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
257 title,
258 token
259 )
260 if isinstance(user_id, Response):
261 return user_id
262
263 # Get Actions table columns names
264 columns: Union[List[str], int] = self.runtime_data_initialised.database_link.get_table_column_names(
265 table_name=CONST.TAB_ACTIONS
266 )
267 if isinstance(columns, int):
268 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(
269 title=title,
270 token=token
271 )
272 columns.pop(0)
273
274 if (isinstance(request_body["trigger"], str) is False):
275 request_body["trigger"] = json.dumps(request_body["trigger"], ensure_ascii=False)
276 if (isinstance(request_body["consequences"], str) is False):
277 request_body["consequences"] = json.dumps(request_body["consequences"], ensure_ascii=False)
278
279 self.disp.log_debug(f"Before inserting in database: {request_body}", title)
280
281 # Insert new applet into database
282 status: int = self.runtime_data_initialised.database_link.insert_data_into_table(
283 table=CONST.TAB_ACTIONS,
284 data=[
285 request_body["name"],
286 request_body["trigger"],
287 request_body["consequences"],
288 user_id,
289 request_body["tags"],
290 "0",
291 request_body["description"],
292 request_body["colour"]
293 ],
294 column=columns
295 )
296 self.disp.log_debug(f"Insert status: {status}", title)
297 if status == self.error:
298 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(
299 title=title,
300 token=token
301 )
302 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
303 title=title,
304 message="The applet was created successfully.",
305 resp="success",
306 token=token
307 )
308 return HCI.success(
309 content=body,
310 content_type=CONST.CONTENT_TYPE,
311 headers=self.runtime_data_initialised.json_header
312 )
313
314 async def put_applet_by_id(self, request: Request, applet_id: str) -> Response:
315 """
316 Modify applet by id
317 """
318 title = "Put applet by id"
319
320 # Token getter
321 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
322 request
323 )
324 if not token:
325 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
326 title
327 )
328 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
329 token
330 ) is False:
331 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
332 title
333 )
334 self.disp.log_debug(f"Token = {token}", title)
335
336 # Request Body getter
337 request_body = await self.runtime_data_initialised.boilerplate_incoming_initialised.get_body(request)
338 self.disp.log_debug(f"Request body: {request_body}", title)
339 if not request_body or not all(key in request_body for key in ("name", "trigger", "consequences", "tags", "description", "colour")):
340 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
341 if request_body["name"] == "" or request_body["description"] == "" or request_body["colour"] == "" or len(request_body["trigger"]) == 0 or len(request_body["consequences"]) == 0:
342 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
343
344 # User id getter
345 user_id: Union[str, Response] = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
346 title,
347 token
348 )
349 if isinstance(user_id, Response):
350 return user_id
351
352 # Verify if the applet is found
353 applet_data = self.runtime_data_initialised.database_link.get_data_from_table(
354 CONST.TAB_ACTIONS,
355 "*",
356 [f"id='{applet_id}'", f"user_id='{user_id}'"]
357 )
358 if not applet_data or isinstance(applet_data, int):
359 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
360 title=title,
361 message="Your applet was not found.",
362 resp="not found",
363 token=token,
364 error=True
365 )
366 return HCI.not_found(
367 content=body,
368 content_type=CONST.CONTENT_TYPE,
369 headers=self.runtime_data_initialised.json_header
370 )
371
372 # Get Actions table columns names
373 columns: Union[List[str], int] = self.runtime_data_initialised.database_link.get_table_column_names(
374 table_name=CONST.TAB_ACTIONS
375 )
376 if isinstance(columns, int):
377 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(
378 title=title,
379 token=token
380 )
381 columns.pop(0)
382
383 if (isinstance(request_body["trigger"], str) is False):
384 request_body["trigger"] = json.dumps(request_body["trigger"], ensure_ascii=False)
385 if (isinstance(request_body["consequences"], str) is False):
386 request_body["consequences"] = json.dumps(request_body["consequences"], ensure_ascii=False)
387
388 # Update applet into database
389 status: int = self.runtime_data_initialised.database_link.update_data_in_table(
390 table=CONST.TAB_ACTIONS,
391 data=[
392 request_body["name"],
393 request_body["trigger"],
394 request_body["consequences"],
395 user_id,
396 request_body["tags"],
397 "0",
398 request_body["description"],
399 request_body["colour"]
400 ],
401 column=columns,
402 where=f"id='{applet_id}'"
403 )
404 if status == self.error:
405 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(
406 title=title,
407 token=token
408 )
409 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
410 title=title,
411 message="The applet was updated successfully.",
412 resp="success",
413 token=token
414 )
415 return HCI.success(
416 content=body,
417 content_type=CONST.CONTENT_TYPE,
418 headers=self.runtime_data_initialised.json_header
419 )
420
421 async def patch_applet_by_id(self, request: Request, applet_id: str) -> Response:
422 """
423 Modify some applet by id
424 """
425 title = "Modify some applet by id"
426
427 # Token getter
428 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
429 request
430 )
431 if not token:
432 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
433 title
434 )
435 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
436 token
437 ) is False:
438 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
439 title
440 )
441 self.disp.log_debug(f"Token = {token}", title)
442
443 # Request Body getter
444 request_body = await self.runtime_data_initialised.boilerplate_incoming_initialised.get_body(request)
445 self.disp.log_debug(f"Request body: {request_body}", title)
446
447 # User id getter
448 user_id: Union[str, Response] = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
449 title,
450 token
451 )
452 if isinstance(user_id, Response):
453 return user_id
454
455 # Verify if the applet is found
456 applet_data = self.runtime_data_initialised.database_link.get_data_from_table(
457 CONST.TAB_ACTIONS,
458 "*",
459 [f"id='{applet_id}'", f"user_id='{user_id}'"]
460 )
461 if not applet_data or isinstance(applet_data, int):
462 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
463 title=title,
464 message="Your applet was not found.",
465 resp="not found",
466 token=token,
467 error=True
468 )
469 return HCI.not_found(
470 content=body,
471 content_type=CONST.CONTENT_TYPE,
472 headers=self.runtime_data_initialised.json_header
473 )
474
475 # Get Actions table columns names
476 columns: Union[List[str], int] = self.runtime_data_initialised.database_link.get_table_column_names(
477 table_name=CONST.TAB_ACTIONS
478 )
479 if isinstance(columns, int):
480 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(
481 title=title,
482 token=token
483 )
484 columns.pop(0)
485
486 # Update applet into database
487 if "name" in request_body:
488 if request_body["name"] == "":
489 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
490 if self.runtime_data_initialised.boilerplate_non_http_initialised.update_single_data(
491 CONST.TAB_ACTIONS,
492 "id",
493 "name",
494 applet_id,
495 request_body
496 ) == self.error:
497 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(title, token)
498 if "trigger" in request_body:
499 if len(request_body["trigger"]) == 0:
500 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
501 if (isinstance(request_body["trigger"], str) is False):
502 request_body["trigger"] = json.dumps(request_body["trigger"], ensure_ascii=False)
503 if self.runtime_data_initialised.boilerplate_non_http_initialised.update_single_data(
504 CONST.TAB_ACTIONS,
505 "id",
506 "trigger",
507 applet_id,
508 request_body
509 ) == self.error:
510 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(title, token)
511 if "consequences" in request_body:
512 if len(request_body["consequences"]) == 0:
513 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
514 if (isinstance(request_body["consequences"], str) is False):
515 request_body["consequences"] = json.dumps(request_body["consequences"], ensure_ascii=False)
516 if self.runtime_data_initialised.boilerplate_non_http_initialised.update_single_data(
517 CONST.TAB_ACTIONS,
518 "id",
519 "consequences",
520 applet_id,
521 request_body
522 ) == self.error:
523 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(title, token)
524 if "tags" in request_body:
525 if self.runtime_data_initialised.boilerplate_non_http_initialised.update_single_data(
526 CONST.TAB_ACTIONS,
527 "id",
528 "tags",
529 applet_id,
530 request_body
531 ) == self.error:
532 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(title, token)
533 if "colour" in request_body:
534 if request_body["colour"] == "":
535 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
536 if self.runtime_data_initialised.boilerplate_non_http_initialised.update_single_data(
537 CONST.TAB_ACTIONS,
538 "id",
539 "colour",
540 applet_id,
541 request_body
542 ) == self.error:
543 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(title, token)
544 if "description" in request_body:
545 if request_body["description"] == "":
546 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(title)
547 if self.runtime_data_initialised.boilerplate_non_http_initialised.update_single_data(
548 CONST.TAB_ACTIONS,
549 "id",
550 "description",
551 applet_id,
552 request_body
553 ) == self.error:
554 return self.runtime_data_initialised.boilerplate_responses_initialised.internal_server_error(title, token)
555 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
556 title=title,
557 message="The applet was updated successfully.",
558 resp="success",
559 token=token
560 )
561 return HCI.success(
562 content=body,
563 content_type=CONST.CONTENT_TYPE,
564 headers=self.runtime_data_initialised.json_header
565 )
566
567 def _search_services(self, type_to_search: str, title: str, token: str) -> Union[List[Dict[str, Any]], Response]:
568 """
569 Search for the services depending on if the type is trigger or action
570 """
571 # Get the data of ActionTemplates table by the triggers type
572 templates_data: Union[List[Dict[str, Any]], int] = self.runtime_data_initialised.database_link.get_data_from_table(
573 CONST.TAB_ACTION_TEMPLATE,
574 "*",
575 f"type='{type_to_search}'"
576 )
577 if not templates_data or isinstance(templates_data, int):
578 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
579 title=title,
580 message="The actions templates were not found.",
581 resp="not found",
582 token=token,
583 error=True
584 )
585 return HCI.not_found(
586 content=body,
587 content_type=CONST.CONTENT_TYPE,
588 headers=self.runtime_data_initialised.json_header
589 )
590
591 # Loop inside the actions templates list
592 already_retrieved_services: List[int] = []
593 services: List[Dict[str, Any]] = []
594 for _, template in enumerate(templates_data):
595 if template["service_id"] in already_retrieved_services:
596 continue
597 key: str = "service_id"
598 service: Union[List[Dict[str, Any]], int] = self.runtime_data_initialised.database_link.get_data_from_table(
599 CONST.TAB_SERVICES,
600 "*",
601 f"id='{template[key]}'"
602 )
603 if not service or isinstance(service, int):
604 continue
605 services.append({
606 "id": service[0]["id"],
607 "name": service[0]["name"],
608 "colour": service[0]["colour"],
609 "description": service[0]["description"]
610 })
611 already_retrieved_services.append(service[0]["id"])
612 return services
613
614 async def get_triggers_services(self, request: Request) -> Response:
615 """
616 Get triggers services
617 """
618 title = "Get triggers services"
619
620 # Token getter
621 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
622 request
623 )
624 if not token:
625 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
626 title
627 )
628 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
629 token
630 ) is False:
631 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
632 title
633 )
634 self.disp.log_debug(f"Token = {token}", title)
635
636 # Get triggers services
637 services: Union[List[Dict[str, Any]], Response] = self._search_services("trigger", title, token)
638 if (isinstance(services, Response)):
639 return services
640
641 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
642 title=title,
643 message=services,
644 resp="success",
645 token=token,
646 )
647 return HCI.success(
648 content=body,
649 content_type=CONST.CONTENT_TYPE,
650 headers=self.runtime_data_initialised.json_header
651 )
652
653 async def get_reactions_services(self, request: Request) -> Response:
654 """
655 Get reactions services
656 """
657 title = "Get reactions services"
658
659 # Token getter
660 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
661 request
662 )
663 if not token:
664 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
665 title
666 )
667 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
668 token
669 ) is False:
670 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
671 title
672 )
673 self.disp.log_debug(f"Token = {token}", title)
674
675 # Get actions services
676 services: Union[List[Dict[str, Any]], Response] = self._search_services("action", title, token)
677 if (isinstance(services, Response)):
678 return services
679
680 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
681 title=title,
682 message=services,
683 resp="success",
684 token=token,
685 )
686 return HCI.success(
687 content=body,
688 content_type=CONST.CONTENT_TYPE,
689 headers=self.runtime_data_initialised.json_header
690 )
691
692 async def get_triggers(self, request: Request) -> Response:
693 """
694 Get triggers by research
695 """
696 title = "Get triggers by research"
697
698 # Token getter
699 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
700 request
701 )
702 if not token:
703 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
704 title
705 )
706 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
707 token
708 ) is False:
709 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
710 title
711 )
712 self.disp.log_debug(f"Token = {token}", title)
713
714 # Get the triggers
715 templates_data: Union[List[Dict[str, Any]], int] = self.runtime_data_initialised.database_link.get_data_from_table(
716 CONST.TAB_ACTION_TEMPLATE,
717 "*",
718 "type='trigger'"
719 )
720 if not templates_data or isinstance(templates_data, int):
721 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
722 title=title,
723 message="The templates were not found.",
724 resp="not found",
725 token=token,
726 error=True
727 )
728 return HCI.not_found(
729 content=body,
730 content_type=CONST.CONTENT_TYPE,
731 headers=self.runtime_data_initialised.json_header
732 )
733
734 for _, template in enumerate(templates_data):
735 template["json"] = json.loads(template["json"])
736 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
737 title=title,
738 message=templates_data,
739 resp="success",
740 token=token
741 )
742 return HCI.success(
743 content=body,
744 content_type=CONST.CONTENT_TYPE,
745 headers=self.runtime_data_initialised.json_header
746 )
747
748 async def get_reactions(self, request: Request) -> Response:
749 """
750 Get triggers by research
751 """
752 title = "Get triggers by research"
753
754 # Token getter
755 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
756 request
757 )
758 if not token:
759 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
760 title
761 )
762 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
763 token
764 ) is False:
765 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
766 title
767 )
768 self.disp.log_debug(f"Token = {token}", title)
769
770 # Get the reactions
771 templates_data: Union[List[Dict[str, Any]], int] = self.runtime_data_initialised.database_link.get_data_from_table(
772 CONST.TAB_ACTION_TEMPLATE,
773 "*",
774 "type='action'"
775 )
776 if not templates_data or isinstance(templates_data, int):
777 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
778 title=title,
779 message="The templates were not found.",
780 resp="not found",
781 token=token,
782 error=True
783 )
784 return HCI.not_found(
785 content=body,
786 content_type=CONST.CONTENT_TYPE,
787 headers=self.runtime_data_initialised.json_header
788 )
789
790 for _, template in enumerate(templates_data):
791 template["json"] = json.loads(template["json"])
792 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
793 title=title,
794 message=templates_data,
795 resp="success",
796 token=token
797 )
798 return HCI.success(
799 content=body,
800 content_type=CONST.CONTENT_TYPE,
801 headers=self.runtime_data_initialised.json_header
802 )
803
804 async def get_triggers_by_service_id(self, request: Request, service_id: str) -> Response:
805 """
806 Get triggers by service name
807 """
808 title = "Get triggers by service name"
809
810 # Token getter
811 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
812 request
813 )
814 if not token:
815 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
816 title
817 )
818 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
819 token
820 ) is False:
821 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
822 title
823 )
824 self.disp.log_debug(f"Token = {token}", title)
825
826 # Get the triggers list by service id from the ActionTemplate table
827 triggers_data = self.runtime_data_initialised.database_link.get_data_from_table(
828 CONST.TAB_ACTION_TEMPLATE,
829 "*",
830 ["type='trigger'", f"service_id='{service_id}'"]
831 )
832 if not triggers_data or isinstance(triggers_data, int):
833 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
834 title=title,
835 message="The triggers were not found.",
836 resp="not found",
837 token=token,
838 error=True
839 )
840 return HCI.not_found(
841 content=body,
842 content_type=CONST.CONTENT_TYPE,
843 headers=self.runtime_data_initialised.json_header
844 )
845
846 for _, trigger in enumerate(triggers_data):
847 trigger["json"] = json.loads(trigger["json"])
848 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
849 title=title,
850 message=triggers_data,
851 resp="success",
852 token=token
853 )
854 return HCI.success(
855 content=body,
856 content_type=CONST.CONTENT_TYPE,
857 headers=self.runtime_data_initialised.json_header
858 )
859
860 async def get_reactions_by_service_id(self, request: Request, service_id: str) -> Response:
861 """
862 Get reactions by service name
863 """
864 title = "Get reactions by service name"
865
866 # Token getter
867 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
868 request
869 )
870 if not token:
871 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
872 title
873 )
874 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
875 token
876 ) is False:
877 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
878 title
879 )
880 self.disp.log_debug(f"Token = {token}", title)
881
882 # Get the reactions list by service id from the ActionTemplate table
883 reactions_data = self.runtime_data_initialised.database_link.get_data_from_table(
884 CONST.TAB_ACTION_TEMPLATE,
885 "*",
886 ["type='action'", f"service_id='{service_id}'"]
887 )
888 if not reactions_data or isinstance(reactions_data, int):
889 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
890 title=title,
891 message="The reactions were not found.",
892 resp="not found",
893 token=token,
894 error=True
895 )
896 return HCI.not_found(
897 content=body,
898 content_type=CONST.CONTENT_TYPE,
899 headers=self.runtime_data_initialised.json_header
900 )
901 for _, reaction in enumerate(reactions_data):
902 reaction["json"] = json.loads(reaction["json"])
903 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
904 title=title,
905 message=reactions_data,
906 resp="success",
907 token=token
908 )
909 return HCI.success(
910 content=body,
911 content_type=CONST.CONTENT_TYPE,
912 headers=self.runtime_data_initialised.json_header
913 )
914
915 async def delete_my_applet_by_id(self, request: Request, applet_id: str) -> Response:
916 """
917 Get reactions by service name
918 """
919 title = "Delete service by id"
920
921 # Token getter
922 token = self.runtime_data_initialised.boilerplate_incoming_initialised.get_token_if_present(
923 request
924 )
925 if not token:
926 return self.runtime_data_initialised.boilerplate_responses_initialised.bad_request(
927 title
928 )
929 if self.runtime_data_initialised.boilerplate_non_http_initialised.is_token_correct(
930 token
931 ) is False:
932 return self.runtime_data_initialised.boilerplate_responses_initialised.invalid_token(
933 title
934 )
935 self.disp.log_debug(f"Token = {token}", title)
936
937 # User id getter
938 user_id: Union[str, Response] = self.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
939 title,
940 token
941 )
942 if isinstance(user_id, Response):
943 return user_id
944
945 # Verify if the applet exist or if the user has the right to delete it
946 applet_data = self.runtime_data_initialised.database_link.get_data_from_table(
947 CONST.TAB_ACTIONS,
948 "*",
949 [f"id='{applet_id}'", f"user_id='{user_id}'"]
950 )
951 if not applet_data or isinstance(applet_data, int):
952 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
953 title=title,
954 message="Your applet was not found or it's not your applet.",
955 resp="not found",
956 token=token,
957 error=True
958 )
959 return HCI.not_found(
960 content=body,
961 content_type=CONST.CONTENT_TYPE,
962 headers=self.runtime_data_initialised.json_header
963 )
964 self.disp.log_debug(f"Applet found: {applet_data}", title)
965 self.runtime_data_initialised.database_link.remove_data_from_table(
966 table=CONST.TAB_ACTIONS,
967 where=f"id='{applet_id}'"
968 )
969 body = self.runtime_data_initialised.boilerplate_responses_initialised.build_response_body(
970 title=title,
971 message="Your applet has been deleted successfully.",
972 resp="success",
973 token=None,
974 )
975 return HCI.success(
976 content=body,
977 content_type=CONST.CONTENT_TYPE,
978 headers=self.runtime_data_initialised.json_header
979 )
Response get_triggers(self, Request request)
Definition applets.py:692
Response get_my_applets_by_tags(self, Request request, str tags)
Definition applets.py:152
Response get_reactions(self, Request request)
Definition applets.py:748
Union[List[Dict[str, Any]], Response] _search_services(self, str type_to_search, str title, str token)
Definition applets.py:567
Response get_reactions_services(self, Request request)
Definition applets.py:653
Response get_my_applets(self, Request request)
Definition applets.py:94
Response get_triggers_by_service_id(self, Request request, str service_id)
Definition applets.py:804
Response get_triggers_services(self, Request request)
Definition applets.py:614
None __init__(self, RuntimeData runtime_data, int success=0, int error=84, bool debug=False)
Definition applets.py:17
Response get_applet_by_id(self, Request request, str applet_id)
Definition applets.py:36
Response put_applet_by_id(self, Request request, str applet_id)
Definition applets.py:314
Response delete_my_applet_by_id(self, Request request, str applet_id)
Definition applets.py:915
Response create_applet(self, Request request)
Definition applets.py:225
Response patch_applet_by_id(self, Request request, str applet_id)
Definition applets.py:421
Response get_reactions_by_service_id(self, Request request, str service_id)
Definition applets.py:860