Coverage for src/melvonaut/api.py: 81%

1from melvonaut.settings import settings 

2import importlib.metadata 

3import psutil 

4from aiohttp import web, hdrs 

5from io import StringIO, BytesIO 

6from aiohttp.web_response import ContentCoding 

7from typing import Callable, Any, Awaitable 

8from melvonaut import utils 

9from shared import constants as con 

10from loguru import logger 

11import asyncio 

12import pathlib 

13import shared.models as models 

14import datetime 

15 

16 

17Handler = Callable[[web.Request], Awaitable[web.StreamResponse]] 

18 

19 

20async def health(request: web.Request) -> web.Response: 

21 """Check if the API is running and healthy. 

22 

23 Args: 

24 request (web.Request): The incoming HTTP request. 

25 

26 Returns: 

27 web.Response: A response with status 200 and text "OK". 

28 """ 

29 return web.Response(status=200, text="OK") 

30 

31 

32async def get_disk_usage(request: web.Request) -> web.Response: 

33 """Retrieve disk usage statistics for root and home directories. 

34 

35 Args: 

36 request (web.Request): The incoming HTTP request. 

37 

38 Returns: 

39 web.Response: JSON response containing disk usage data. 

40 """ 

41 logger.debug("Getting disk usage") 

42 disk_root = psutil.disk_usage("/") 

43 disk_home = psutil.disk_usage("/home") 

44 return web.json_response( 

45 {"root": disk_root._asdict(), "home": disk_home._asdict()}, status=200 

46 ) 

47 

48 

49async def get_memory_usage(request: web.Request) -> web.Response: 

50 """Retrieve memory usage statistics. 

51 

52 Args: 

53 request (web.Request): The incoming HTTP request. 

54 

55 Returns: 

56 web.Response: JSON response containing memory usage data. 

57 """ 

58 logger.debug("Getting memory usage") 

59 memory_usage = psutil.virtual_memory() 

60 return web.json_response(memory_usage._asdict(), status=200) 

61 

62 

63async def get_cpu_usage(request: web.Request) -> web.Response: 

64 """Retrieve CPU usage statistics. 

65 

66 Args: 

67 request (web.Request): The incoming HTTP request. 

68 

69 Returns: 

70 web.Response: JSON response containing CPU usage data including user, system, idle time, percent usage, core count, and frequency. 

71 """ 

72 logger.debug("Getting CPU usage") 

73 cpu_usage = psutil.cpu_times() 

74 cpu_percent = psutil.cpu_percent() 

75 cpu_count = psutil.cpu_count() 

76 cpu_freq = psutil.cpu_freq() 

77 cpu = { 

78 "user": cpu_usage.user, 

79 "system": cpu_usage.system, 

80 "idle": cpu_usage.idle, 

81 "percent": cpu_percent, 

82 "physical_cores": cpu_count, 

83 "current_freq": cpu_freq.current, 

84 "max_freq": cpu_freq.max, 

85 "min_freq": cpu_freq.min, 

86 } 

87 return web.json_response(cpu, status=200) 

88 

89 

90async def get_restart_melvin(request: web.Request) -> web.Response: 

91 """Handles a request to restart the Melvin service. 

92 

93 This endpoint is not yet implemented and always returns a 501 Not Implemented response. 

94 

95 Args: 

96 request (web.Request): The incoming HTTP request. 

97 

98 Returns: 

99 web.Response: A response indicating that the operation is not implemented. 

100 """ 

101 return web.Response(status=501, text="Not Implemented") 

102 

103 

104async def get_shutdown_melvin(request: web.Request) -> web.Response: 

105 """Handles a request to shut down the Melvin service. 

106 

107 If `settings.DO_ACTUALLY_EXIT` is set to True, the event loop is stopped, 

108 all pending tasks are canceled, and the process exits. Otherwise, a warning is logged. 

109 

110 Args: 

111 request (web.Request): The incoming HTTP request. 

112 

113 Returns: 

114 web.Response: A response with status 200 indicating the shutdown request was received. 

115 """ 

116 try: 

117 return web.Response(status=200, text="OK") 

118 finally: 

119 if settings.DO_ACTUALLY_EXIT: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 loop = asyncio.get_running_loop() 

121 loop.stop() 

122 pending_tasks = asyncio.all_tasks() 

123 for task in pending_tasks: 

124 task.cancel() 

125 exit() 

126 else: 

127 logger.warning("Requested shutdown, but not actually exiting") 

128 

129 

130async def post_execute_command(request: web.Request) -> web.Response: 

131 """Execute a shell command asynchronously. 

132 

133 Args: 

134 request (web.Request): The incoming HTTP request containing JSON data with the command to execute. 

135 

136 Returns: 

137 web.Response: JSON response containing command output and return code. 

138 """ 

139 data = await request.json() 

140 cmd = data.get("cmd") 

141 logger.debug(f"Executing command: {cmd}") 

142 output = [] 

143 try: 

144 process = await asyncio.create_subprocess_shell( 

145 cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE 

146 ) 

147 while True: 

148 if process.stdout: 148 ↛ 147line 148 didn't jump to line 147 because the condition on line 148 was always true

149 line = await process.stdout.readline() 

150 if not line: 

151 break 

152 output.append(line.decode()) 

153 await process.wait() 

154 return_code = process.returncode 

155 logger.debug(f"Command output: {output}") 

156 return web.json_response( 

157 {"output": output, "return_code": return_code}, status=200 

158 ) 

159 except asyncio.TimeoutError: 

160 return web.json_response({"output": output, "error": "Timeout"}, status=500) 

161 except asyncio.CancelledError: 

162 return web.json_response({"output": output, "error": "Cancelled"}, status=500) 

163 except Exception as e: 

164 return web.json_response({"output": output, "error": str(e)}, status=500) 

165 

166 

167async def get_melvin_version(request: web.Request) -> web.Response: 

168 """Retrieve the current version of the Melvin service. 

169 

170 Args: 

171 request (web.Request): The incoming HTTP request. 

172 

173 Returns: 

174 web.Response: JSON response containing the Melvin service version. 

175 """ 

176 return web.json_response( 

177 {"version": importlib.metadata.version("ciarc")}, status=200 

178 ) 

179 

180 

181async def get_list_log_files(request: web.Request) -> web.Response: 

182 """Retrieve a list of log files from the log directory. 

183 

184 Args: 

185 request (web.Request): The incoming HTTP request. 

186 

187 Returns: 

188 web.Response: JSON response containing a list of log filenames. 

189 """ 

190 logger.debug("Listing log files") 

191 log_files = [] 

192 folder = pathlib.Path(con.MEL_LOG_PATH) 

193 try: 

194 for file in folder.iterdir(): 

195 if not file.is_file(): 

196 continue 

197 if not file.name.endswith(".log"): 

198 continue 

199 log_files.append(file.name) 

200 except Exception as e: 

201 return web.json_response({"error": str(e)}, status=500) 

202 return web.json_response({"log_files": log_files}, status=200) 

203 

204 

205async def post_download_log(request: web.Request) -> web.Response | web.FileResponse: 

206 """Handles log file download requests. 

207 

208 Args: 

209 request (web.Request): The incoming HTTP request containing the log file name in JSON format. 

210 

211 Returns: 

212 web.Response: The requested log file if it exists, otherwise a 404 response. 

213 """ 

214 data = await request.json() 

215 logger.debug(f"Downloading log: {data}") 

216 log_file = pathlib.Path(con.MEL_LOG_PATH) / data.get("file") 

217 if log_file.exists(): 217 ↛ 220line 217 didn't jump to line 220 because the condition on line 217 was always true

218 return web.FileResponse(log_file, status=200) 

219 else: 

220 return web.Response(status=404, text="File not found") 

221 

222 

223async def post_download_log_and_clear(request: web.Request) -> web.Response: 

224 """Handles log file download requests and deletes the file after serving it. 

225 

226 Args: 

227 request (web.Request): The incoming HTTP request containing the log file name in JSON format. 

228 

229 Returns: 

230 web.Response: The requested log file if it exists, otherwise a 404 response. 

231 """ 

232 data = await request.json() 

233 logger.debug(f"Downloading log and clearing: {data}") 

234 log_file = pathlib.Path(con.MEL_LOG_PATH) / data.get("file") 

235 if log_file.exists(): 235 ↛ 243line 235 didn't jump to line 243 because the condition on line 235 was always true

236 log_file_content = StringIO(log_file.read_text()) 

237 try: 

238 return web.Response(body=log_file_content, status=200) 

239 finally: 

240 log_file.unlink() 

241 utils.setup_file_logging() 

242 else: 

243 return web.Response(status=404, text="File not found") 

244 

245 

246async def post_clear_log(request: web.Request) -> web.Response: 

247 """Handles log file deletion requests. 

248 

249 Args: 

250 request (web.Request): The incoming HTTP request containing the log file name in JSON format. 

251 

252 Returns: 

253 web.Response: A success response if the log file is cleared, otherwise an error response. 

254 """ 

255 data = await request.json() 

256 logger.debug(f"Clearing log: {data}") 

257 log_file = pathlib.Path(con.MEL_LOG_PATH) / data.get("file") 

258 if log_file.exists(): 258 ↛ 267line 258 didn't jump to line 267 because the condition on line 258 was always true

259 if log_file.is_dir(): 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true

260 return web.Response(status=400, text=f"{log_file} is a directory") 

261 if not log_file.name.endswith(".log"): 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true

262 return web.Response(status=400, text=f"{log_file} is not a log file") 

263 log_file.unlink() 

264 utils.setup_file_logging() 

265 return web.Response(status=200, text=f"{log_file} cleared") 

266 else: 

267 return web.Response(status=404, text=f"{log_file} not found") 

268 

269 

270async def get_clear_all_logs(request: web.Request) -> web.Response: 

271 """Clears all log files in the system. 

272 

273 Args: 

274 request (web.Request): The incoming HTTP request. 

275 

276 Returns: 

277 web.Response: JSON response containing a list of cleared log files. 

278 """ 

279 try: 

280 logger.debug("Clearing all log files") 

281 log_files = [] 

282 folder = pathlib.Path(con.MEL_LOG_PATH) 

283 for file in folder.iterdir(): 

284 if file.is_dir(): 

285 continue 

286 if not file.name.endswith(".log"): 

287 continue 

288 log_files.append(file.name) 

289 file.unlink() 

290 utils.setup_file_logging() 

291 return web.json_response({"Cleared_files": log_files}, status=200) 

292 except Exception as e: 

293 logger.exception(e) 

294 return web.json_response({"error": str(e)}, status=500) 

295 

296 

297# Download telemetry 

298async def get_download_telemetry( 

299 request: web.Request, 

300) -> web.Response | web.FileResponse: 

301 """Handles telemetry data download requests. 

302 

303 Args: 

304 request (web.Request): The incoming HTTP request. 

305 

306 Returns: 

307 web.Response: The telemetry file if it exists, otherwise a 404 response. 

308 """ 

309 logger.debug("Downloading telemetry") 

310 telemetry_file = pathlib.Path(con.TELEMETRY_LOCATION_CSV) 

311 if telemetry_file.exists(): 

312 return web.FileResponse(telemetry_file, status=200) 

313 else: 

314 return web.Response(status=404, text="File not found") 

315 

316 

317async def get_download_telemetry_and_clear(request: web.Request) -> web.Response: 

318 """Handles telemetry data download requests and deletes the file after serving it. 

319 

320 Args: 

321 request (web.Request): The incoming HTTP request. 

322 

323 Returns: 

324 web.Response: The telemetry file if it exists, otherwise a 404 response. 

325 """ 

326 logger.debug("Downloading telemetry and clearing") 

327 telemetry_file = pathlib.Path(con.TELEMETRY_LOCATION_CSV) 

328 if telemetry_file.exists(): 328 ↛ 335line 328 didn't jump to line 335 because the condition on line 328 was always true

329 telemetry_file_content = StringIO(telemetry_file.read_text()) 

330 try: 

331 return web.Response(body=telemetry_file_content, status=200) 

332 finally: 

333 telemetry_file.unlink() 

334 else: 

335 return web.Response(status=404, text="File not found") 

336 

337 

338async def get_clear_telemetry(request: web.Request) -> web.Response: 

339 """Clears the telemetry data file. 

340 

341 Args: 

342 request (web.Request): The incoming HTTP request. 

343 

344 Returns: 

345 web.Response: A success response if the file is deleted, otherwise a 404 response. 

346 """ 

347 logger.debug("Clearing telemetry") 

348 telemetry_file = pathlib.Path(con.TELEMETRY_LOCATION_CSV) 

349 if telemetry_file.exists(): 349 ↛ 353line 349 didn't jump to line 353 because the condition on line 349 was always true

350 telemetry_file.unlink() 

351 return web.Response(status=200, text="OK") 

352 else: 

353 return web.Response(status=404, text="File not found") 

354 

355 

356# Download events 

357async def get_download_events(request: web.Request) -> web.Response | web.FileResponse: 

358 """Handles the download of event logs. 

359 

360 Args: 

361 request (web.Request): The incoming HTTP request. 

362 

363 Returns: 

364 web.FileResponse: The requested event log file if it exists. 

365 web.Response: A 404 response if the file is not found. 

366 """ 

367 logger.debug("Downloading events") 

368 events_file = pathlib.Path(con.EVENT_LOCATION_CSV) 

369 if events_file.exists(): 

370 return web.FileResponse(events_file, status=200) 

371 else: 

372 return web.Response(status=404, text="File not found") 

373 

374 

375async def get_download_events_and_clear(request: web.Request) -> web.Response: 

376 """Downloads and clears the event log file. 

377 

378 This function retrieves the event log file, sends its content as a response, 

379 and then deletes the file from the system. 

380 

381 Args: 

382 request (web.Request): The incoming HTTP request. 

383 

384 Returns: 

385 web.Response: A response containing the event log content if found. 

386 web.Response: A 404 response if the file is not found. 

387 """ 

388 logger.debug("Downloading events and clearing") 

389 events_file = pathlib.Path(con.EVENT_LOCATION_CSV) 

390 if events_file.exists(): 390 ↛ 397line 390 didn't jump to line 397 because the condition on line 390 was always true

391 events_file_content = StringIO(events_file.read_text()) 

392 try: 

393 return web.Response(body=events_file_content, status=200) 

394 finally: 

395 events_file.unlink() 

396 else: 

397 return web.Response(status=404, text="File not found") 

398 

399 

400async def get_clear_events(request: web.Request) -> web.Response: 

401 """Deletes the event log file from the system. 

402 

403 If the event log file exists, it is deleted. If it does not exist, a 404 response is returned. 

404 

405 Args: 

406 request (web.Request): The incoming HTTP request. 

407 

408 Returns: 

409 web.Response: A 200 response if the file is successfully deleted. 

410 web.Response: A 404 response if the file is not found. 

411 """ 

412 logger.debug("Clearing events") 

413 events_file = pathlib.Path(con.EVENT_LOCATION_CSV) 

414 if events_file.exists(): 414 ↛ 418line 414 didn't jump to line 418 because the condition on line 414 was always true

415 events_file.unlink() 

416 return web.Response(status=200, text="OK") 

417 else: 

418 return web.Response(status=404, text="File not found") 

419 

420 

421async def get_list_images(request: web.Request) -> web.Response: 

422 """Lists all available image files. 

423 

424 Args: 

425 request (web.Request): The incoming HTTP request. 

426 

427 Returns: 

428 web.Response: JSON response containing a list of image filenames. 

429 """ 

430 logger.debug("Listing images") 

431 folder = pathlib.Path(con.IMAGE_PATH_BASE) 

432 if not folder.exists(): 432 ↛ 433line 432 didn't jump to line 433 because the condition on line 432 was never true

433 return web.Response(status=404, text=f"Folder not found: {folder}") 

434 images = [str(file.name) for file in folder.rglob("*.png") if file.is_file()] 

435 return web.json_response({"images": images}, status=200) 

436 

437 

438async def post_download_image(request: web.Request) -> web.Response | web.FileResponse: 

439 """Handles image file download requests. 

440 

441 Args: 

442 request (web.Request): The incoming HTTP request containing the image filename in JSON format. 

443 

444 Returns: 

445 web.Response: The requested image file if it exists, otherwise a 404 response. 

446 """ 

447 data = await request.json() 

448 logger.debug(f"Downloading image: {data}") 

449 image_file = pathlib.Path(con.IMAGE_PATH_BASE) / data.get("file") 

450 if image_file.exists(): 450 ↛ 453line 450 didn't jump to line 453 because the condition on line 450 was always true

451 return web.FileResponse(image_file, status=200) 

452 else: 

453 return web.Response(status=404, text="File not found") 

454 

455 

456async def post_download_image_and_clear(request: web.Request) -> web.Response: 

457 """Handles image file download requests and deletes the file after serving it. 

458 

459 Args: 

460 request (web.Request): The incoming HTTP request containing the image filename in JSON format. 

461 

462 Returns: 

463 web.Response: The requested image file if it exists, otherwise a 404 response. 

464 """ 

465 data = await request.json() 

466 logger.debug(f"Downloading image and clearing: {data}") 

467 image_file = pathlib.Path(con.IMAGE_PATH_BASE) / data.get("file") 

468 if image_file.exists(): 468 ↛ 475line 468 didn't jump to line 475 because the condition on line 468 was always true

469 image_file_content = BytesIO(image_file.read_bytes()) 

470 try: 

471 return web.Response(body=image_file_content, status=200) 

472 finally: 

473 image_file.unlink() 

474 else: 

475 return web.Response(status=404, text="File not found") 

476 

477 

478async def get_clear_all_images(request: web.Request) -> web.Response: 

479 """Clears all stored images. 

480 

481 Args: 

482 request (web.Request): The incoming HTTP request. 

483 

484 Returns: 

485 web.Response: JSON response containing a list of cleared images. 

486 """ 

487 logger.debug("Clearing all images") 

488 folder = pathlib.Path(con.IMAGE_PATH_BASE) 

489 images = [str(file) for file in folder.rglob("*.png") if file.is_file()] 

490 for image in images: 

491 pathlib.Path(image).unlink() 

492 return web.Response(status=200, text="OK") 

493 

494 

495async def post_set_melvin_task(request: web.Request) -> web.Response: 

496 """Sets a task for Melvin (a task management system). 

497 

498 Args: 

499 request (web.Request): The incoming HTTP request containing the task details in JSON format. 

500 

501 Returns: 

502 web.Response: A success response if the task is set. 

503 """ 

504 data = await request.json() 

505 logger.debug(f"Setting melvin task: {data}") 

506 task = data.get("task", None) 

507 if not task: 507 ↛ 508line 507 didn't jump to line 508 because the condition on line 507 was never true

508 logger.warning("Missing field task") 

509 return web.Response(status=400, text="Missing field task") 

510 try: 

511 melvin_task = models.MELVINTask(task) 

512 except ValueError: 

513 logger.warning("Invalid task") 

514 return web.Response(status=400, text="Invalid task") 

515 except Exception as e: 

516 logger.warning(f"Error setting melvin task: {e}") 

517 return web.Response(status=500, text=str(e)) 

518 settings.set_settings({"CURRENT_MELVIN_TASK": melvin_task}) 

519 return web.Response(status=200, text="OK") 

520 

521 

522async def get_reset_settings(request: web.Request) -> web.Response: 

523 """Resets all settings to their default values. 

524 

525 Args: 

526 request (web.Request): The incoming HTTP request. 

527 

528 Returns: 

529 web.Response: A success response confirming the reset. 

530 """ 

531 logger.debug("Resetting settings") 

532 settings.clear_settings() 

533 return web.Response(status=200, text="OK") 

534 

535 

536async def post_set_setting(request: web.Request) -> web.Response: 

537 """Sets a new configuration setting. 

538 

539 Args: 

540 request (web.Request): The incoming HTTP request containing settings in JSON format. 

541 

542 Returns: 

543 web.Response: A success response if the setting is applied. 

544 """ 

545 logger.debug("Setting settings") 

546 data = await request.json() 

547 logger.debug(f"Setting settings: {data}") 

548 settings.set_settings(data) 

549 return web.Response(status=200, text="OK") 

550 

551 

552async def post_clear_setting(request: web.Request) -> web.Response: 

553 """Clears a specific setting. 

554 

555 Args: 

556 request (web.Request): The incoming HTTP request containing the setting name in JSON format. 

557 

558 Returns: 

559 web.Response: A success response if the setting is cleared. 

560 """ 

561 logger.debug("Clearing settings") 

562 data = await request.json() 

563 logger.debug(f"Clearing settings: {data}") 

564 settings.delete_settings(data.keys()) 

565 return web.Response(status=200, text="OK") 

566 

567 

568async def post_get_setting(request: web.Request) -> web.Response: 

569 """Retrieves a specific setting. 

570 

571 Args: 

572 request (web.Request): The incoming HTTP request containing the setting name in JSON format. 

573 

574 Returns: 

575 web.Response: JSON response containing the requested setting value. 

576 """ 

577 logger.debug("Getting settings") 

578 data = await request.json() 

579 logger.debug(f"Requested settings: {data}") 

580 response_settings = {} 

581 try: 

582 for key in data.keys(): 

583 response_settings[key] = settings.__getattribute__(key) 

584 except AttributeError: 

585 return web.Response(status=404, text=f"Setting not found: {key}") 

586 logger.debug(f"Response settings: {response_settings}") 

587 return web.json_response(response_settings, status=200) 

588 

589 

590async def get_all_settings(request: web.Request) -> web.Response: 

591 """Retrieve all settings configured in the system. 

592 

593 Args: 

594 request (web.Request): The incoming HTTP request. 

595 

596 Returns: 

597 web.Response: JSON response containing all system settings. 

598 """ 

599 logger.debug("Getting all settings") 

600 attrs = [key for key in dir(settings) if not key.startswith("_") and key.isupper()] 

601 all_settings: dict[str, Any] = {} 

602 for attr in attrs: 

603 value = settings.__getattribute__(attr) 

604 if type(value) is float or type(value) is int: 

605 all_settings[attr] = value 

606 elif type(value) is datetime.datetime: 

607 all_settings[attr] = value.isoformat() 

608 else: 

609 all_settings[attr] = str(value) 

610 return web.json_response(all_settings, status=200) 

611 

612 

613def setup_routes(app: web.Application) -> None: 

614 """Sets up API routes for the web application. 

615 

616 Args: 

617 app (web.Application): The web application instance. 

618 """ 

619 app.router.add_post("/api/post_download_log", post_download_log) 

620 app.router.add_get("/api/get_download_telemetry", get_download_telemetry) 

621 app.router.add_get("/api/get_download_events", get_download_events) 

622 app.router.add_post("/api/post_download_image", post_download_image) 

623 app.router.add_post("/api/post_set_melvin_task", post_set_melvin_task) 

624 app.router.add_get("/api/get_reset_settings", get_reset_settings) 

625 app.router.add_post("/api/post_set_setting", post_set_setting) 

626 app.router.add_post("/api/post_clear_setting", post_clear_setting) 

627 app.router.add_post("/api/post_clear_log", post_clear_log) 

628 app.router.add_post("/api/post_get_setting", post_get_setting) 

629 app.router.add_get("/api/get_all_settings", get_all_settings) 

630 app.router.add_post("/api/post_download_log_and_clear", post_download_log_and_clear) 

631 app.router.add_get( 

632 "/api/get_download_telemetry_and_clear", get_download_telemetry_and_clear 

633 ) 

634 app.router.add_get( 

635 "/api/get_download_events_and_clear", get_download_events_and_clear 

636 ) 

637 app.router.add_post( 

638 "/api/post_download_image_and_clear", post_download_image_and_clear 

639 ) 

640 app.router.add_get("/api/get_clear_all_logs", get_clear_all_logs) 

641 app.router.add_get("/api/get_clear_telemetry", get_clear_telemetry) 

642 app.router.add_get("/api/get_clear_events", get_clear_events) 

643 app.router.add_get("/api/get_clear_all_images", get_clear_all_images) 

644 app.router.add_get("/api/health", health) 

645 app.router.add_get("/api/get_disk_usage", get_disk_usage) 

646 app.router.add_get("/api/get_memory_usage", get_memory_usage) 

647 app.router.add_get("/api/get_cpu_usage", get_cpu_usage) 

648 app.router.add_get("/api/get_restart_melvin", get_restart_melvin) 

649 app.router.add_get("/api/get_shutdown_melvin", get_shutdown_melvin) 

650 app.router.add_post("/api/post_execute_command", post_execute_command) 

651 app.router.add_get("/api/get_melvin_version", get_melvin_version) 

652 app.router.add_get("/api/get_list_log_files", get_list_log_files) 

653 app.router.add_get("/api/get_list_images", get_list_images) 

654 

655 

656@web.middleware 

657async def compression_middleware(request: web.Request, handler: Handler) -> Any: 

658 """Middleware to handle response compression using gzip or deflate. 

659 

660 This middleware checks the `Accept-Encoding` header of the request 

661 to determine if the client supports gzip or deflate compression. 

662 If supported, it applies the corresponding compression to the response. 

663 

664 Args: 

665 request (web.Request): The incoming HTTP request. 

666 handler (Handler): The next request handler in the middleware chain. 

667 

668 Returns: 

669 Any: The compressed HTTP response if the client supports it, 

670 otherwise the original response. 

671 """ 

672 accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower() 

673 

674 if ContentCoding.gzip.value in accept_encoding: 

675 compressor = ContentCoding.gzip.value 

676 elif ContentCoding.deflate.value in accept_encoding: 676 ↛ 679line 676 didn't jump to line 679 because the condition on line 676 was always true

677 compressor = ContentCoding.deflate.value 

678 else: 

679 return await handler(request) 

680 

681 resp = await handler(request) 

682 resp.headers[hdrs.CONTENT_ENCODING] = compressor 

683 resp.enable_compression() 

684 return resp 

685 

686 

687@web.middleware 

688async def catcher_middleware(request: web.Request, handler: Handler) -> Any: 

689 """Middleware to catch and log unhandled exceptions in request handling. 

690 

691 If an exception occurs while processing the request, this middleware 

692 logs the error and returns a 500 Internal Server Error response 

693 with the error message. 

694 

695 Args: 

696 request (web.Request): The incoming HTTP request. 

697 handler (Handler): The next request handler in the middleware chain. 

698 

699 Returns: 

700 Any: The HTTP response from the handler, or a 500 error response 

701 if an exception occurs. 

702 """ 

703 try: 

704 return await handler(request) 

705 except Exception as e: 

706 logger.exception(e) 

707 return web.Response(status=500, text=str(e)) 

708 

709 

710async def run_api() -> None: 

711 """Starts the web API server. 

712 

713 Returns: 

714 None 

715 """ 

716 logger.debug("Setting up API server") 

717 settings.init_settings() 

718 app = web.Application(middlewares=[compression_middleware, catcher_middleware]) 

719 setup_routes(app) 

720 runner = web.AppRunner(app) 

721 await runner.setup() 

722 site = web.TCPSite(runner, "0.0.0.0", settings.API_PORT) 

723 try: 

724 logger.info(f"API server started on port {settings.API_PORT}") 

725 await site.start() 

726 logger.debug("API server started") 

727 finally: 

728 # das hat bei mir direkt nach dem start ausgelöst 

729 # logger.debug("Shutting down API server") 

730 # await runner.cleanup() 

731 pass