w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-09-20 14:27:42 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Extract webserver metrics into a separate class to avoid passing it into every API call. Change the display to requests per hour rather than just last request time. Add SSE and telnet client connected count.

Browse Source
This commit is contained in:
Ian Renton
2026-09-11 22:32:04 +01:00
parent 4a09e46fe0
commit 7c458a8c5b
25 changed files with 161 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webserver/handlers/api/addspot.py
+1 -12
View File
@@ -1,10 +1,8 @@
import logging
import re
import threading
from datetime import datetime
from typing import Any
import pytz
import requests
import tornado
from tornado import httputil
@@ -12,7 +10,6 @@ from tornado.web import Application
from core.config import ALLOW_SPOTTING, ALLOW_UPSTREAM_SPOTTING, RECAPTCHA_SECRET_KEY
from core.constants import UNKNOWN_BAND
from core.prometheus_metrics_handler import api_requests_counter
from core.sig_utils import get_ref_regex_for_sig
from core.utils import infer_band_from_freq, safe_json_dumps
from data.spot import Spot
@@ -33,23 +30,15 @@ class APISpotHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._spots = None
self._web_server_metrics = None
self._spot_providers = None
super().__init__(application, request, **kwargs)
def initialize(self, spots, web_server_metrics, spot_providers=None):
def initialize(self, spots, spot_providers=None):
self._spots = spots
self._web_server_metrics = web_server_metrics
self._spot_providers = spot_providers or []
def post(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# Reject if not allowed
if not ALLOW_SPOTTING:
self.set_status(401)
webserver/handlers/api/alerts.py
+2 -19
View File
@@ -10,7 +10,6 @@ from tornado import httputil
from tornado.web import Application
from core.enums import AlertType
from core.prometheus_metrics_handler import api_requests_counter
from core.utils import safe_json_dumps
from data.lookup_credentials import extract_credentials
@@ -27,12 +26,10 @@ class APIAlertsHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._alerts = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, alerts, web_server_metrics):
def initialize(self, alerts):
self._alerts = alerts
self._web_server_metrics = web_server_metrics
@staticmethod
def _enrich(alerts, credentials):
@@ -45,12 +42,6 @@ class APIAlertsHandler(tornado.web.RequestHandler):
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
@@ -82,15 +73,13 @@ class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
def __init__(self, application, request, **kwargs: Any):
self._sse_alert_broadcaster = None
self._web_server_metrics = None
self._query_params = None
self._credentials = None
self._fields = None
super().__init__(application, request, **kwargs)
def initialize(self, sse_alert_broadcaster, web_server_metrics):
def initialize(self, sse_alert_broadcaster):
self._sse_alert_broadcaster = sse_alert_broadcaster
self._web_server_metrics = web_server_metrics
def custom_headers(self):
"""Custom headers to avoid e.g. nginx reverse proxy from buffering SSE data"""
@@ -99,12 +88,6 @@ class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
def open(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
webserver/handlers/api/dxstats.py
+1 -9
View File
@@ -11,7 +11,6 @@ from tornado.web import Application
from core.constants import BANDS
from core.enums import Continent
from core.prometheus_metrics_handler import api_requests_counter
from core.utils import safe_json_dumps
logger = logging.getLogger(__name__)
@@ -30,20 +29,13 @@ class APIDxStatsHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._spots = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, spots, web_server_metrics):
def initialize(self, spots):
self._spots = spots
self._web_server_metrics = web_server_metrics
def get(self):
try:
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
one_hour_ago = (datetime.now(pytz.UTC) - timedelta(hours=1)).timestamp()
counts = Counter()
webserver/handlers/api/lookups.py
-33
View File
@@ -1,9 +1,7 @@
import logging
import re
from datetime import datetime
from typing import Any
import pytz
import tornado
from tornado import httputil
from tornado.web import Application
@@ -15,7 +13,6 @@ from core.geo_utils import (
lat_lon_to_cq_zone,
lat_lon_to_itu_zone,
)
from core.prometheus_metrics_handler import api_requests_counter
from core.sig_lookup_helper import populate_missing_sig_ref_info
from core.sig_utils import get_ref_regex_for_sig
from core.utils import safe_json_dumps
@@ -34,20 +31,10 @@ class APILookupCallHandler(tornado.web.RequestHandler):
request: httputil.HTTPServerRequest,
**kwargs: Any,
):
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, web_server_metrics):
self._web_server_metrics = web_server_metrics
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
@@ -85,20 +72,10 @@ class APILookupSIGRefHandler(tornado.web.RequestHandler):
request: httputil.HTTPServerRequest,
**kwargs: Any,
):
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, web_server_metrics):
self._web_server_metrics = web_server_metrics
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
@@ -143,20 +120,10 @@ class APILookupGridHandler(tornado.web.RequestHandler):
request: httputil.HTTPServerRequest,
**kwargs: Any,
):
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, web_server_metrics):
self._web_server_metrics = web_server_metrics
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
webserver/handlers/api/options.py
+1 -12
View File
@@ -1,8 +1,6 @@
import logging
from datetime import datetime
from typing import Any
import pytz
import tornado
from tornado import httputil
from tornado.web import Application
@@ -10,7 +8,6 @@ from tornado.web import Application
from core.config import ALLOW_SPOTTING, MAX_SPOT_AGE
from core.constants import BANDS, PROPAGATION_MODES, SIGS
from core.enums import Continent, Mode, ModeType
from core.prometheus_metrics_handler import api_requests_counter
from core.utils import safe_json_dumps
logger = logging.getLogger(__name__)
@@ -26,23 +23,15 @@ class APIOptionsHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._status_data = None
self._web_server_metrics = None
self._spot_providers = None
super().__init__(application, request, **kwargs)
def initialize(self, status_data, web_server_metrics, spot_providers=None):
def initialize(self, status_data, spot_providers=None):
self._status_data = status_data
self._web_server_metrics = web_server_metrics
self._spot_providers = spot_providers or []
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# Build a map of SIG name -> list of provider names that can submit spots for that SIG
spot_submit_providers = {}
webserver/handlers/api/solar_conditions.py
+1 -12
View File
@@ -1,13 +1,10 @@
import logging
from datetime import datetime
from typing import Any
import pytz
import tornado
from tornado import httputil
from tornado.web import Application
from core.prometheus_metrics_handler import api_requests_counter
from core.utils import safe_json_dumps
logger = logging.getLogger(__name__)
@@ -23,21 +20,13 @@ class APISolarConditionsHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._solar_conditions = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, solar_conditions, web_server_metrics):
def initialize(self, solar_conditions):
self._solar_conditions = solar_conditions
self._web_server_metrics = web_server_metrics
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
self.write(self._solar_conditions.to_json())
self.set_status(200)
self.set_header("Cache-Control", "no-store")
webserver/handlers/api/spots.py
+2 -19
View File
@@ -9,7 +9,6 @@ import tornado_eventsource.handler
from tornado import httputil
from tornado.web import Application
from core.prometheus_metrics_handler import api_requests_counter
from core.utils import safe_json_dumps
from data.lookup_credentials import extract_credentials
@@ -26,12 +25,10 @@ class APISpotsHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._spots = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, spots, web_server_metrics):
def initialize(self, spots):
self._spots = spots
self._web_server_metrics = web_server_metrics
@staticmethod
def _enrich(spots, credentials):
@@ -44,12 +41,6 @@ class APISpotsHandler(tornado.web.RequestHandler):
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
@@ -81,15 +72,13 @@ class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
def __init__(self, application, request, **kwargs: Any):
self._sse_spot_broadcaster = None
self._web_server_metrics = None
self._query_params = None
self._credentials = None
self._fields = None
super().__init__(application, request, **kwargs)
def initialize(self, sse_spot_broadcaster, web_server_metrics):
def initialize(self, sse_spot_broadcaster):
self._sse_spot_broadcaster = sse_spot_broadcaster
self._web_server_metrics = web_server_metrics
def custom_headers(self):
"""Custom headers to avoid e.g. nginx reverse proxy from buffering SSE data"""
@@ -100,12 +89,6 @@ class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
"""Called once on the client opening a connection, set things up"""
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# request.arguments contains lists for each param key because technically the client can supply multiple,
# reduce that to just the first entry, and convert bytes to string
self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
webserver/handlers/api/status.py
+1 -12
View File
@@ -1,13 +1,10 @@
import logging
from datetime import datetime
from typing import Any
import pytz
import tornado
from tornado import httputil
from tornado.web import Application
from core.prometheus_metrics_handler import api_requests_counter
from core.utils import safe_json_dumps
logger = logging.getLogger(__name__)
@@ -23,21 +20,13 @@ class APIStatusHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._status_data = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, status_data, web_server_metrics):
def initialize(self, status_data):
self._status_data = status_data
self._web_server_metrics = web_server_metrics
def get(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
self.write(safe_json_dumps(self._status_data))
self.set_status(200)
self.set_header("Cache-Control", "no-store")
webserver/handlers/api/v1_addspot.py
+1 -12
View File
@@ -1,16 +1,13 @@
import logging
import re
from datetime import datetime
from typing import Any
import pytz
import tornado
from tornado import httputil
from tornado.web import Application
from core.config import ALLOW_SPOTTING
from core.constants import UNKNOWN_BAND
from core.prometheus_metrics_handler import api_requests_counter
from core.sig_utils import get_ref_regex_for_sig
from core.utils import infer_band_from_freq, safe_json_dumps
from data.spot import Spot
@@ -28,21 +25,13 @@ class V1APISpotHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._spots = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, spots, web_server_metrics):
def initialize(self, spots):
self._spots = spots
self._web_server_metrics = web_server_metrics
def post(self):
try:
# Metrics
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["api_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
api_requests_counter.inc()
# Reject if not allowed
if not ALLOW_SPOTTING:
self.set_status(401)
webserver/handlers/pagetemplate.py
+1 -12
View File
@@ -1,7 +1,5 @@
from datetime import datetime
from typing import Any
import pytz
import tornado
from tornado import httputil
from tornado.web import Application
@@ -16,7 +14,6 @@ from core.config import (
WEB_UI_OPTIONS,
)
from core.constants import SOFTWARE_VERSION
from core.prometheus_metrics_handler import page_requests_counter
class PageTemplateHandler(tornado.web.RequestHandler):
@@ -29,20 +26,12 @@ class PageTemplateHandler(tornado.web.RequestHandler):
**kwargs: Any,
):
self._template_name = None
self._web_server_metrics = None
super().__init__(application, request, **kwargs)
def initialize(self, template_name, web_server_metrics):
def initialize(self, template_name):
self._template_name = template_name
self._web_server_metrics = web_server_metrics
def get(self):
# Metrics
self._web_server_metrics["last_page_access_time"] = datetime.now(pytz.UTC)
self._web_server_metrics["page_access_counter"] += 1
self._web_server_metrics["status"] = "OK"
page_requests_counter.inc()
# Load named template, and provide variables used in templates
self.render(
f"{self._template_name}.html",
webserver/sse_broadcaster.py
+5
View File
@@ -26,6 +26,11 @@ class SSEBroadcaster:
with self._lock:
self._handlers.discard(handler)
@property
def client_count(self) -> int:
with self._lock:
return len(self._handlers)
def publish(self, value):
self._loop.add_callback(self._broadcast, value)
webserver/webserver.py
+36 -37
View File
@@ -33,6 +33,7 @@ from webserver.handlers.manifesthandler import ManifestHandler
from webserver.handlers.metrics import PrometheusMetricsHandler
from webserver.handlers.pagetemplate import PageTemplateHandler
from webserver.sse_broadcaster import SSEBroadcaster
from webserver.webserver_metrics import WebServerMetrics
logger = logging.getLogger(__name__)
@@ -52,19 +53,19 @@ class WebServer:
self._port = WEB_SERVER_PORT
self._api_only_mode = API_ONLY_MODE
self._shutdown_event = asyncio.Event()
self.web_server_metrics = {
"last_page_access_time": None,
"last_api_access_time": None,
"page_access_counter": 0,
"api_access_counter": 0,
"status": "Starting",
}
self.web_server_metrics = WebServerMetrics()
def setup(self):
# Listen for new spots and alerts being added to the cache, so we can notify SSE clients immediately
DATA_STORE.spots.add_listener(self._spot_broadcaster.publish)
DATA_STORE.alerts.add_listener(self._alert_broadcaster.publish)
@property
def sse_client_count(self) -> int:
"""Number of connected SSE clients, across both the spots and alerts streams."""
return self._spot_broadcaster.client_count + self._alert_broadcaster.client_count
def start(self):
"""Start the web server"""
@@ -82,62 +83,57 @@ class WebServer:
self._spot_broadcaster.bind_to_web_server_loop()
self._alert_broadcaster.bind_to_web_server_loop()
# Prepare a list of common arguments that are passed in to every API & page handler. This is just a basic thing
# to avoid copy-pasting the same thing to every route declaration below.
handler_opts = {"web_server_metrics": self.web_server_metrics}
# API endpoints are always enabled
api_routes = [
(
r"/api/v2/spots",
APISpotsHandler,
{"spots": self._data_store.spots, **handler_opts},
{"spots": self._data_store.spots},
),
(
r"/api/v2/alerts",
APIAlertsHandler,
{"alerts": self._data_store.alerts, **handler_opts},
{"alerts": self._data_store.alerts},
),
(
r"/api/v2/spots/stream",
APISpotsStreamHandler,
{"sse_spot_broadcaster": self._spot_broadcaster, **handler_opts},
{"sse_spot_broadcaster": self._spot_broadcaster},
),
(
r"/api/v2/alerts/stream",
APIAlertsStreamHandler,
{"sse_alert_broadcaster": self._alert_broadcaster, **handler_opts},
{"sse_alert_broadcaster": self._alert_broadcaster},
),
(
r"/api/v2/solar",
APISolarConditionsHandler,
{"solar_conditions": self._data_store.solar_conditions.get(), **handler_opts},
{"solar_conditions": self._data_store.solar_conditions.get()},
),
(
r"/api/v2/dxstats",
APIDxStatsHandler,
{"spots": self._data_store.spots, **handler_opts},
{"spots": self._data_store.spots},
),
(
r"/api/v2/options",
APIOptionsHandler,
{"status_data": self._data_store.status.get(), **handler_opts},
{"status_data": self._data_store.status.get()},
),
(
r"/api/v2/status",
APIStatusHandler,
{"status_data": self._data_store.status.get(), **handler_opts},
{"status_data": self._data_store.status.get()},
),
(r"/api/v2/lookup/call", APILookupCallHandler, {**handler_opts}),
(r"/api/v2/lookup/sigref", APILookupSIGRefHandler, {**handler_opts}),
(r"/api/v2/lookup/grid", APILookupGridHandler, {**handler_opts}),
(r"/api/v2/lookup/call", APILookupCallHandler),
(r"/api/v2/lookup/sigref", APILookupSIGRefHandler),
(r"/api/v2/lookup/grid", APILookupGridHandler),
(
r"/api/v2/spot",
APISpotHandler,
{
"spots": self._data_store.spots,
"spot_providers": self._data_providers,
**handler_opts,
},
),
]
@@ -148,19 +144,18 @@ class WebServer:
(
r"/api/v1/spots",
V1APISpotsHandler,
{"spots": self._data_store.spots, **handler_opts},
{"spots": self._data_store.spots},
),
(
r"/api/v1/spots/stream",
V1APISpotsStreamHandler,
{"sse_spot_broadcaster": self._spot_broadcaster, **handler_opts},
{"sse_spot_broadcaster": self._spot_broadcaster},
),
(
r"/api/v1/spot",
V1APISpotHandler,
{
"spots": self._data_store.spots,
**handler_opts,
},
),
(r"/api/v1/(.*)", V1RedirectHandler),
@@ -173,41 +168,41 @@ class WebServer:
(
r"/",
PageTemplateHandler,
{"template_name": "api_only_home", **handler_opts},
{"template_name": "api_only_home"},
)
]
else:
ui_routes = [
(r"/", PageTemplateHandler, {"template_name": "spots", **handler_opts}),
(r"/", PageTemplateHandler, {"template_name": "spots"}),
(
r"/map",
PageTemplateHandler,
{"template_name": "map", **handler_opts},
{"template_name": "map"},
),
(
r"/bands",
PageTemplateHandler,
{"template_name": "bands", **handler_opts},
{"template_name": "bands"},
),
(
r"/alerts",
PageTemplateHandler,
{"template_name": "alerts", **handler_opts},
{"template_name": "alerts"},
),
(
r"/conditions",
PageTemplateHandler,
{"template_name": "conditions", **handler_opts},
{"template_name": "conditions"},
),
(
r"/status",
PageTemplateHandler,
{"template_name": "status", **handler_opts},
{"template_name": "status"},
),
(
r"/about",
PageTemplateHandler,
{"template_name": "about", **handler_opts},
{"template_name": "about"},
),
]
# Only allow the Add Spot page if spotting is allowed
@@ -216,7 +211,7 @@ class WebServer:
(
r"/add-spot",
PageTemplateHandler,
{"template_name": "add_spot", **handler_opts},
{"template_name": "add_spot"},
)
]
@@ -226,7 +221,7 @@ class WebServer:
(
r"/apidocs",
PageTemplateHandler,
{"template_name": "apidocs", **handler_opts},
{"template_name": "apidocs"},
),
(r"/metrics", PrometheusMetricsHandler),
(r"/manifest.webmanifest", ManifestHandler),
@@ -254,7 +249,11 @@ class WebServer:
def request_log(handler):
"""Custom log function to provide more data about requests when enabled, and to provide the ability to turn off
web request logging altogetether."""
web request logging altogetether. Also records the time of the request and status in the webserver metrics. Probably
not what this method is supposed to be used for but it's a convenient thing that gets called on every request, so
saves having to pass the metrics around each handler individually."""
WEB_SERVER.web_server_metrics.record(handler.request.path, handler.get_status())
if LOG_WEB_REQUESTS:
if handler.get_status() < 500:
webserver/webserver_metrics.py
+44
View File
@@ -0,0 +1,44 @@
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)