152 lines
6.2 KiB
Python
152 lines
6.2 KiB
Python
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
import logging
|
|
import os
|
|
import json
|
|
import hashlib
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import List, Optional, Dict, Any
|
|
from datetime import datetime
|
|
from http import HTTPStatus
|
|
import re
|
|
|
|
from alarm_storage import AlarmStorage
|
|
from data_classes import RepeatRule, Snooze, Metadata, Alarm
|
|
from logging_config import setup_logging
|
|
|
|
# Set up logging configuration
|
|
logger = setup_logging()
|
|
|
|
class AlertApi(BaseHTTPRequestHandler):
|
|
def __init__(self, *args, **kwargs):
|
|
self.storage = AlarmStorage("data/alerts.json")
|
|
self.logger = logger
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def _send_response(self, status_code: int, data: Any = None, error: str = None) -> None:
|
|
"""Send a JSON response with the given status code and data/error"""
|
|
self.send_response(status_code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
|
|
response = {}
|
|
if data is not None:
|
|
response["data"] = data
|
|
if error is not None:
|
|
response["error"] = error
|
|
|
|
response_json = json.dumps(response)
|
|
self.logger.debug(f"Sending response: {response_json}")
|
|
self.wfile.write(response_json.encode("utf-8"))
|
|
|
|
def _handle_request(self, method: str) -> None:
|
|
"""Handle incoming requests with proper error handling"""
|
|
self.logger.info(f"Received {method} request from {self.client_address[0]}")
|
|
|
|
try:
|
|
if method in ["POST", "PUT", "DELETE"]:
|
|
content_length = int(self.headers.get("Content-Length", 0))
|
|
if content_length == 0:
|
|
raise ValueError("Missing request body")
|
|
|
|
post_data = self.rfile.read(content_length)
|
|
self.logger.debug(f"Received {method} payload: {post_data.decode('utf-8')}")
|
|
|
|
post_data = json.loads(post_data)
|
|
else:
|
|
post_data = None
|
|
|
|
# Route request to appropriate handler
|
|
handler = getattr(self, f"_handle_{method.lower()}", None)
|
|
if handler:
|
|
self.logger.debug(f"Routing to handler: _handle_{method.lower()}")
|
|
handler(post_data)
|
|
else:
|
|
self.logger.warning(f"Method not allowed: {method}")
|
|
self._send_response(HTTPStatus.METHOD_NOT_ALLOWED, error="Method not allowed")
|
|
|
|
except json.JSONDecodeError as e:
|
|
self.logger.error(f"JSON decode error: {str(e)}")
|
|
self._send_response(HTTPStatus.BAD_REQUEST, error="Invalid JSON in request body")
|
|
except ValueError as e:
|
|
self.logger.error(f"Validation error: {str(e)}")
|
|
self._send_response(HTTPStatus.BAD_REQUEST, error=str(e))
|
|
except Exception as e:
|
|
self.logger.error(f"Unexpected error: {str(e)}", exc_info=True)
|
|
self._send_response(HTTPStatus.INTERNAL_SERVER_ERROR, error="Internal server error")
|
|
|
|
def _handle_get(self, _) -> None:
|
|
"""Handle GET request"""
|
|
self.logger.debug("Processing GET request for all alarms")
|
|
alarms = self.storage.get_saved_alerts()
|
|
self.logger.debug(f"Retrieved {len(alarms)} alarms")
|
|
self._send_response(HTTPStatus.OK, data=alarms)
|
|
|
|
def _handle_post(self, data: dict) -> None:
|
|
"""Handle POST request"""
|
|
self.logger.debug(f"Processing POST request with data: {json.dumps(data, indent=2)}")
|
|
try:
|
|
alarm_id = self.storage.save_new_alert(data)
|
|
self.logger.info(f"Successfully created new alarm with ID: {alarm_id}")
|
|
self._send_response(HTTPStatus.CREATED, data={"id": alarm_id})
|
|
except ValueError as e:
|
|
self.logger.error(f"Failed to create alarm: {str(e)}")
|
|
self._send_response(HTTPStatus.BAD_REQUEST, error=str(e))
|
|
|
|
def _handle_put(self, data: dict) -> None:
|
|
"""Handle PUT request"""
|
|
alarm_id = data.pop("id", None)
|
|
self.logger.debug(f"Processing PUT request for alarm ID {alarm_id} with data: {json.dumps(data, indent=2)}")
|
|
|
|
if alarm_id is None:
|
|
self.logger.error("PUT request missing alarm ID")
|
|
self._send_response(HTTPStatus.BAD_REQUEST, error="Missing alarm ID")
|
|
return
|
|
|
|
if self.storage.update_alert(alarm_id, data):
|
|
self.logger.info(f"Successfully updated alarm ID: {alarm_id}")
|
|
self._send_response(HTTPStatus.OK, data={"message": "Alarm updated successfully"})
|
|
else:
|
|
self.logger.warning(f"Alarm not found for update: {alarm_id}")
|
|
self._send_response(HTTPStatus.NOT_FOUND, error="Alarm not found")
|
|
|
|
def _handle_delete(self, data: dict) -> None:
|
|
"""Handle DELETE request"""
|
|
alarm_id = data.get("id")
|
|
self.logger.debug(f"Processing DELETE request for alarm ID: {alarm_id}")
|
|
|
|
if not isinstance(alarm_id, int):
|
|
self.logger.error(f"Invalid alarm ID format: {alarm_id}")
|
|
self._send_response(HTTPStatus.BAD_REQUEST, error="Invalid alarm ID")
|
|
return
|
|
|
|
if self.storage.remove_saved_alert(alarm_id):
|
|
self.logger.info(f"Successfully deleted alarm ID: {alarm_id}")
|
|
self._send_response(HTTPStatus.OK, data={"message": "Alarm removed successfully"})
|
|
else:
|
|
self.logger.warning(f"Alarm not found for deletion: {alarm_id}")
|
|
self._send_response(HTTPStatus.NOT_FOUND, error="Alarm not found")
|
|
|
|
def do_GET(self): self._handle_request("GET")
|
|
def do_POST(self): self._handle_request("POST")
|
|
def do_PUT(self): self._handle_request("PUT")
|
|
def do_DELETE(self): self._handle_request("DELETE")
|
|
|
|
|
|
def run(server_class=HTTPServer, handler_class=AlertApi, port=8000):
|
|
logger.info(f"Starting AlertApi on port {port}")
|
|
|
|
server_address = ("", port)
|
|
httpd = server_class(server_address, handler_class)
|
|
|
|
try:
|
|
logger.info("Server is ready to handle requests")
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
logger.info("Received shutdown signal")
|
|
except Exception as e:
|
|
logger.error(f"Server error: {str(e)}", exc_info=True)
|
|
finally:
|
|
httpd.server_close()
|
|
logger.info("Server stopped")
|