Coverage for yaptide/celery/utils/requests.py: 67%

1import logging 

2import os 

3 

4import requests 

5 

6 

7def send_task_update(simulation_id: int, task_id: int, update_key: str, update_dict: dict) -> bool: 

8 """Sends task status to backend which will update the database""" 

9 flask_url = os.environ.get("BACKEND_INTERNAL_URL") 

10 if not flask_url: 

11 logging.warning("Flask URL not found via BACKEND_INTERNAL_URL") 

12 return False 

13 if not update_key: 

14 logging.warning("Update key not found, skipping update") 

15 return False 

16 dict_to_send = { 

17 "simulation_id": simulation_id, 

18 "task_id": task_id, 

19 "update_key": update_key, 

20 "update_dict": update_dict 

21 } 

22 logging.debug("Sending update %s to the backend %s", dict_to_send, flask_url) 

23 res: requests.Response = requests.Session().post(url=f"{flask_url}/tasks", json=dict_to_send) 

24 if res.status_code != 202: 

25 logging.warning("Update_dict: %s", update_dict) 

26 logging.warning("Task update for %s - Failed: %s", task_id, res.json().get("message")) 

27 return False 

28 return True 

29 

30 

31def send_simulation_results(simulation_id: int, update_key: str, estimators: list) -> bool: 

32 """Sends simulation results to flask to save it in database""" 

33 flask_url = os.environ.get("BACKEND_INTERNAL_URL") 

34 if not flask_url: 

35 logging.warning("Flask URL not found via BACKEND_INTERNAL_URL") 

36 return False 

37 if not update_key: 

38 logging.warning("Update key not found, skipping update") 

39 return False 

40 dict_to_send = { 

41 "simulation_id": simulation_id, 

42 "update_key": update_key, 

43 "estimators": estimators, 

44 } 

45 logging.info("Sending results to flask via %s", flask_url) 

46 res: requests.Response = requests.Session().post(url=f"{flask_url}/results", json=dict_to_send) 

47 if res.status_code != 202: 

48 logging.warning("Saving results failed: %s", res.json().get("message")) 

49 return False 

50 return True 

51 

52 

53def send_simulation_logfiles(simulation_id: int, update_key: str, logfiles: dict) -> bool: 

54 """ 

55 Sends simulation logfiles to Flask backend which will save it in database 

56 Returns True if successful, False otherwise 

57 """ 

58 flask_url = os.environ.get("BACKEND_INTERNAL_URL") 

59 if not flask_url: 

60 logging.warning("Flask URL not found via BACKEND_INTERNAL_URL") 

61 return False 

62 dict_to_send = { 

63 "simulation_id": simulation_id, 

64 "update_key": update_key, 

65 "logfiles": logfiles, 

66 } 

67 logging.info("Sending log files to flask via %s", flask_url) 

68 res: requests.Response = requests.Session().post(url=f"{flask_url}/logfiles", json=dict_to_send) 

69 if res.status_code != 202: 

70 logging.warning("Saving logfiles failed: %s", res.json()["message"]) 

71 return False 

72 return True