Coverage for app/routes/logging.py: 81%
72 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-02 02:49 +0000
1from flask import request, Blueprint
2from app.services.log_service import log_event, get_all_logs, get_logs_by_user, log_suggestion, get_ai_usage, get_logs_by_class
3from flasgger import swag_from
4from app.models.response import *
5from app.models.status_codes import StatusCodes
9logging_bp = Blueprint('logging', __name__)
12@logging_bp.route('/logs', methods=['POST'])
13@swag_from({
14 'tags': ['Logging'],
15 'summary': 'Log an event',
16 'description': 'Logs the event to the database.',
17 'parameters': [
18 {
19 'name': 'body',
20 'in': 'body',
21 'required': True,
22 'schema': {
23 'type': 'object',
24 'properties': {
25 'event': {
26 'type': 'string',
27 'example': 'User logged in'
28 },
29 'metadata': {
30 'type': 'object',
31 'example': {
32 'userID': 12345,
33 'time_lapse': 1708
34 }
35 }
36 },
37 'required': ['event', 'metadata']
38 }
39 }
40 ],
41 'responses': {
42 '200': {
43 'description': 'Event logged successfully',
44 'schema': {
45 'type': 'object',
46 'properties': {
47 'status': {'type': 'string', 'example': 'logged'}
48 }
49 }
50 },
51 '400': {
52 'description': 'Bad request or invalid input',
53 'schema': {
54 'type': 'object',
55 'properties': {
56 'status': {'type': 'string', 'example': 'error'},
57 'message': {'type': 'string', 'example': 'Missing required fields: event'}
58 }
59 }
60 },
61 '500': {
62 'description': 'Internal server error'
63 }
64 }
65})
66def log_event_route():
67 """
68 Logs the event to the database.
69 See Swagger docs for more information.
70 """
71 data = request.json
73 required_fields = ['event', 'time_lapse', 'metadata']
74 missing_fields = [field for field in required_fields if field not in data]
76 if missing_fields:
77 return error_response(
78 f"Missing required fields: {', '.join(missing_fields)}",
79 None,
80 StatusCodes.BAD_REQUEST
81 )
83 try:
84 log_event(data)
85 return success_response(
86 "Logged event",
87 None,
88 StatusCodes.CREATED
89 )
91 except Exception as e:
92 return error_response(
93 f"Error logging event: {e}",
94 None,
95 StatusCodes.SERVER_ERROR
96 )
99@logging_bp.route('/logs', methods=['GET'])
100@swag_from({
101 'tags': ['Logging'],
102 'summary': 'Retrieve all logs',
103 'description': 'Fetches all logged events from the database.',
104 'responses': {
105 '200': {
106 'description': 'List of logs',
107 'schema': {
108 'type': 'array',
109 'items': {
110 'type': 'object',
111 'properties': {
112 'id': {'type': 'integer', 'example': 1},
113 'timestamp': {'type': 'number', 'example': 1708401940},
114 'event': {'type': 'string', 'example': 'user_login'},
115 'data': {'type': 'object', 'example': {'user_id': 12345}}
116 }
117 }
118 }
119 },
120 '500': {
121 'description': 'Internal server error'
122 }
123 }
124})
125def get_all_logs_route():
126 """
127 Retrieve all logs in the database
128 See Swagger docs for more information.
129 """
130 try:
131 logs = get_all_logs()
132 return success_response(
133 "All logs",
134 logs,
135 StatusCodes.OK
136 )
138 except Exception as e:
139 return error_response(
140 f"Error fetching logs: {e}",
141 None,
142 StatusCodes.SERVER_ERROR
143 )
146@logging_bp.route('/logs/<string:user_id>', methods=['GET'])
147@swag_from({
148 'tags': ['Logging'],
149 'summary': 'TODO Retrieve logs by user ID',
150 'description': 'Fetches all logged events associated with a specific user ID.',
151 'parameters': [
152 {
153 'name': 'user_id',
154 'in': 'path',
155 'required': True,
156 'type': 'string',
157 'description': 'The ID of the user whose logs are being retrieved.',
158 'example': 12345
159 }
160 ],
161 'responses': {
162 '200': {
163 'description': 'List of logs for the specified user',
164 'schema': {
165 'type': 'array',
166 'items': {
167 'type': 'object',
168 'properties': {
169 'id': {'type': 'integer', 'example': 1},
170 'timestamp': {'type': 'number', 'example': 1708401940},
171 'event': {'type': 'string', 'example': 'user_login'},
172 'data': {'type': 'object', 'example': {'user_id': 12345}}
173 }
174 }
175 }
176 },
177 '404': {
178 'description': 'No logs found for the specified user'
179 },
180 '500': {
181 'description': 'Internal server error'
182 }
183 }
184})
185def get_logs_by_user_route(user_id):
186 """
187 Get all logs for a specific user
188 See Swagger docs for more information.
189 """
190 try:
191 user_section_id = request.args.get("user_section_id")
192 user_class_id = request.args.get("user_class_id")
194 logs = get_logs_by_user(user_id, user_section_id=user_section_id, user_class_id=user_class_id)
196 if not logs:
197 return success_response(
198 f"No logs found for user {user_id}",
199 )
201 return success_response(
202 f"Logs for user {user_id}",
203 logs,
204 )
205 except Exception as e:
206 return error_response(
207 f"Error fetching logs for user {user_id}: {e}",
208 None,
209 StatusCodes.SERVER_ERROR
210 )
212@logging_bp.route("/logs/class/<string:class_id>", methods=["GET"])
213def get_logs_by_class_route(class_id):
214 """
215 Get all logs for a specific class (user_class_id)
216 """
217 try:
218 logs = get_logs_by_class(class_id)
220 if not logs:
221 return success_response(
222 f"No logs found for class {class_id}",
223 []
224 )
226 return success_response(
227 f"Logs for class {class_id}",
228 logs,
229 )
230 except Exception as e:
231 return error_response(
232 f"Error fetching logs for class {class_id}: {e}",
233 None,
234 StatusCodes.SERVER_ERROR
235 )
239@logging_bp.route('/logs/suggestion', methods=['POST'])
240@swag_from({
241 'tags': ['Logging'],
242 'summary': 'Log a suggestion',
243 'description': 'Logs a suggestion to the database.',
244 'parameters': [
245 {
246 'name': 'body',
247 'in': 'body',
248 'required': True,
249 'schema': {
250 'type': 'object',
251 'properties': {
252 'prompt': {
253 'type': 'string',
254 'example': 'function add(a, b)'
255 },
256 'suggestionText': {
257 'type': 'string',
258 'example': '{\n return a + b;\n}'
259 },
260 'hasBug': {
261 'type': 'boolean',
262 'example': False
263 },
264 'model': {
265 'type': 'string',
266 'example': 'gpt-3'
267 },
268 'userSectionid': {
269 'type': 'string',
270 'example': '12345'
271 },
272 },
273 'required': ['prompt', 'suggestionText', 'hasBug', 'model']
274 }
275 }
276 ],
277 'responses': {
278 '200': {
279 'description': 'Suggestion logged successfully',
280 'schema': {
281 'type': 'object',
282 'properties': {
283 'status': {'type': 'string', 'example': 'success'},
284 'data': {
285 'type': 'object',
286 'properties': {
287 'id': {'type': 'string', 'example': "12345"},
288 'prompt': {'type': 'string', 'example': 'function add(a, b)'},
289 'suggestionText': {'type': 'string', 'example': '{\n return a + b;\n}'},
290 'hasBug': {'type': 'boolean', 'example': False},
291 'model': {'type': 'string', 'example': 'gpt-3'},
292 'created_at': {'type': 'timestamp', 'example': '2023-10-01T12:00:00Z'}
293 }
294 }
295 }
296 }
297 },
298 '400': {
299 'description': 'Bad request or invalid input',
300 'schema': {
301 'type': 'object',
302 'properties': {
303 'status': {'type': 'string', 'example': 'error'},
304 'message': {'type': 'string', 'example': 'Missing required fields: prompt, suggestionText'}
305 }
306 }
307 },
308 '500': {
309 'description': 'Internal server error',
310 'schema': {
311 'type': 'object',
312 'properties': {
313 'status': {'type': 'string', 'example': 'error'},
314 'message': {'type': 'string', 'example': 'Internal server error'}
315 }
316 }
317 }
318 }
319})
320def log_suggestion_route():
321 data = request.json
323 # Validate required fields
324 required_fields = ['prompt', 'suggestionArray', 'hasBug', 'model', 'userSectionId']
325 missing_fields = [field for field in required_fields if field not in data]
327 if missing_fields:
328 return error_response(
329 f"Missing required fields: {', '.join(missing_fields)}",
330 None,
331 StatusCodes.BAD_REQUEST
332 )
334 suggestion = {
335 'prompt': data['prompt'],
336 'suggestion_array': data['suggestionArray'],
337 'has_bug': data['hasBug'],
338 'model': data['model'],
339 'user_section_id': data.get('userSectionId')
340 }
342 try:
343 logged_suggestion = log_suggestion(suggestion)
345 return success_response(
346 "Logged suggestion",
347 logged_suggestion['id'],
348 StatusCodes.CREATED
349 )
351 except Exception as e:
352 return error_response(
353 f"Error logging event: {e}",
354 None,
355 StatusCodes.SERVER_ERROR
356 )
358@logging_bp.route('/logs/ai', methods=['GET'])
359def ai_usage_route():
360 """
361 Retrieve AI usage statistics
362 See Swagger docs for more information.
363 """
364 try:
365 ai_usage = get_ai_usage()
367 return success_response(
368 "AI usage statistics",
369 ai_usage,
370 StatusCodes.OK
371 )
373 except Exception as e:
374 return error_response(
375 f"Error fetching AI usage: {e}",
376 None,
377 StatusCodes.SERVER_ERROR
378 )
380@logging_bp.route('/health', methods=['GET'])
381def health_check():
382 """
383 Health check endpoint to verify the service is running.
384 """
385 return success_response(
386 "Service is running",
387 None,
388 StatusCodes.OK
389 )