48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import date
|
|
from http import HTTPStatus
|
|
from urllib.parse import parse_qs
|
|
|
|
|
|
class AlertRoutesMixin:
|
|
def _handle_alerts_get(self, parsed) -> bool:
|
|
if parsed.path == "/api/alerts":
|
|
query = parse_qs(parsed.query)
|
|
try:
|
|
self.send_json(
|
|
self.application_service.alert_center(
|
|
query.get("status", ["all"])[0],
|
|
query.get("as_of", [date.today().isoformat()])[0],
|
|
)
|
|
)
|
|
except ValueError as exc:
|
|
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
|
|
return True
|
|
return False
|
|
|
|
def _handle_alerts_post(self, parsed) -> bool:
|
|
alert_read_match = re.fullmatch(r"/api/alerts/(\d+)/read", parsed.path)
|
|
if alert_read_match:
|
|
self.send_json(
|
|
{"ok": True, **self.application_service.mark_alert_read(int(alert_read_match.group(1)))}
|
|
)
|
|
return True
|
|
if parsed.path == "/api/alerts/read-all":
|
|
body = self.read_json_body(True)
|
|
self.send_json(
|
|
{"ok": True, **self.application_service.mark_all_alerts_read(str(body.get("as_of") or ""))}
|
|
)
|
|
return True
|
|
return False
|
|
|
|
def _handle_alerts_delete(self, parsed) -> bool:
|
|
alert_match = re.fullmatch(r"/api/alerts/(\d+)", parsed.path)
|
|
if alert_match:
|
|
self.send_json(
|
|
{"ok": True, **self.application_service.delete_alert(int(alert_match.group(1)))}
|
|
)
|
|
return True
|
|
return False
|