from collections import deque from datetime import datetime, timedelta import pytz from core.prometheus_metrics_handler import api_requests_counter, page_requests_counter class WebServerMetrics: """Tracker for web server metrics. Stores the times pages and API endpoints were accessed for an hour, so we can display the rate of requests per hour, and also updates the equivalent Prometheus counters.""" def __init__(self): self.status = "Starting" self._page_access_times = deque() self._api_access_times = deque() def record(self, path: str, status_code: int): """Records data for a request, depending on whether it's a page, API, or other request, and making sure the response code isn't 404. Also sets the status of the web server.""" if status_code == 404 or path.startswith(("/static/", "/metrics", "/manifest.webmanifest")): return self.status = "OK" if status_code < 500 else "Error" if path.startswith("/api/"): api_requests_counter.inc() self._api_access_times.append(datetime.now(pytz.UTC)) else: page_requests_counter.inc() self._page_access_times.append(datetime.now(pytz.UTC)) def page_requests_per_hour(self) -> int: return self._count_and_prune_last_hour(self._page_access_times) def api_requests_per_hour(self) -> int: return self._count_and_prune_last_hour(self._api_access_times) @staticmethod def _count_and_prune_last_hour(access_times: deque) -> int: cutoff = datetime.now(pytz.UTC) - timedelta(hours=1) while access_times and access_times[0] < cutoff: access_times.popleft() return len(access_times)