Coverage for app/services/log_service.py: 72%

1from app.controllers.database import client 

2from app.models.response import * 

3 

4 

5 

6def log_event(event): 

7 """Logs an event to the database.""" 

8 try: 

9 client.table("logs").insert(event).execute() 

10 

11 except Exception as e: 

12 print(f"Error logging event: {e}") 

13 raise e 

14 

15 

16def log_suggestion(suggestion): 

17 """ 

18 Logs an event by inserting it into the 'Logs' table in the database. 

19 

20 Args: 

21 event (dict): A dictionary containing event details. Expected keys are: 

22 - 'event': The name of the event. 

23 - 'time_lapse': A textual description of the event. 

24 - 'metadata': Additional data associated with the event. 

25 

26 Raises: 

27 Exception: If there is an error inserting the log into the database. 

28 """ 

29 try: 

30 response = client.table("suggestions").insert(suggestion).execute() 

31 if response.data: 

32 suggestion['id'] = response.data[0]['id'] 

33 return suggestion 

34 else: 

35 raise Exception("No data returned from insert operation") 

36 except Exception as e: 

37 print(f"Error logging suggestion: {e}") 

38 raise e 

39 

40def get_all_logs(): 

41 """ 

42 Retrieves all logs stored in the 'Logs' table. 

43 

44 Returns: 

45 list: A list of dictionaries containing all logs. 

46 

47 Raises: 

48 Exception: If there is an error fetching the logs from the database. 

49 """ 

50 try: 

51 response = client.table("logs").select("*").execute() 

52 return response.data 

53 except Exception as e: 

54 print(f"Error fetching logs: {e}") 

55 raise e 

56 

57def get_logs_by_user(user_id, user_section_id=None, user_class_id=None): 

58 """ 

59 Retrieves all logs associated with a specific user. 

60 

61 Args: 

62 user_id (str): The ID of the user whose logs are to be fetched. 

63 

64 Returns: 

65 list: A list of dictionaries containing logs for the specified user. 

66 

67 Raises: 

68 Exception: If there is an error fetching logs for the user. 

69 """ 

70 try: 

71 query = client.table("logs").select("*").eq("metadata->>user_id", str(user_id)) 

72 

73 if user_section_id is not None: 

74 query = query.eq("metadata->>user_section_id", str(user_section_id)) 

75 

76 if user_class_id is not None: 

77 query = query.eq("metadata->>user_class_id", str(user_class_id)) 

78 

79 response = query.execute() 

80 

81 return response.data 

82 except Exception as e: 

83 print(f"Error fetching logs for user {user_id}: {e}") 

84 raise e 

85 

86 

87def get_ai_usage(): 

88 """ 

89 Retrieves AI usage statistics from the database. 

90 

91 Returns: 

92 dict: A dictionary containing AI usage statistics. 

93 

94 Raises: 

95 Exception: If there is an error fetching AI usage data from the database. 

96 """ 

97 try: 

98 response = client.table("ai_usage").select("*").execute() 

99 return response.data 

100 except Exception as e: 

101 print(f"Error fetching AI usage data: {e}") 

102 raise e 

103 

104def get_all_data_from_db(): 

105 """ 

106 Retrieves all data from the database. 

107 

108 Returns: 

109 list: A list of dictionaries containing all data from the database. 

110 

111 Raises: 

112 Exception: If there is an error fetching data from the database. 

113 """ 

114 try: 

115 users = client.table("users").select("*").execute() 

116 class_users = client.table("class_users").select("*").execute() 

117 classes = client.table("classes").select("*").execute() 

118 logs = client.table("logs").select("*").execute() 

119 suggestion = client.table("suggestions").select("*").execute() 

120 user_section_questions = client.table("user_section_questions").select("*").execute() 

121 user_sections = client.table("user_sections").select("*").execute() 

122 

123 response = { 

124 "users": users.data, 

125 "class_users": class_users.data, 

126 "classes": classes.data, 

127 "logs": logs.data, 

128 "suggestion": suggestion.data, 

129 "user_section_questions": user_section_questions.data, 

130 "user_sections": user_sections.data 

131 } 

132 return response 

133 except Exception as e: 

134 print(f"Error fetching all data: {e}") 

135 raise e 

136 

137def get_logs_by_class(class_id): 

138 """ 

139 Retrieves all logs for users in a specific class. 

140 

141 Args: 

142 class_id (str): The class ID to filter logs by. 

143 

144 Returns: 

145 list: A list of logs matching the class ID. 

146 """ 

147 try: 

148 query = client.table("logs").select("*").eq("metadata->>user_class_id", str(class_id)) 

149 response = query.execute() 

150 

151 if not response.data: 

152 return [] 

153 

154 return response.data 

155 except Exception as e: 

156 print(f"Error fetching logs for class {class_id}: {e}") 

157 raise e