refactor: move /api/stats into core/stats_routes.py (#1563)

The sample extraction from Discussion #1536: `/api/stats` moves out of
`main.py` into `shelfmark/core/stats_routes.py`, registered with
`register_stats_routes(app, ...)` like the other route modules. The
route logic is the same, and no test changed.

`main.py` passes two callables instead of values, one for the user
database and one for the auth mode. They are called per request, so the
tests that patch `main.user_db` and `main.get_auth_mode` after import
still work. The read-only key allowlist in the middleware now uses the
path constant from the new module, so the path lives in one place.

This is only the shape. If it looks right I'll move the route groups
from the 1.5.0 PRs the same way, one PR each. If you'd rather leave
`main.py` as it is, close this and no harm done.
This commit is contained in:
splitsec2 authored and GitHub committed 2026-10-23 21:51:58 -06:00
1 parent 61419451e4
commit 6b775cd9e8
2 files changed
+62 -21

No files matched your search

+55
View File
@@ -0,0 +1,55 @@
"""Dashboard counters: ``GET /api/stats``."""
from __future__ import annotations
import sqlite3
from typing import TYPE_CHECKING
from flask import g, jsonify, session
from shelfmark.core import stats
from shelfmark.core.logger import setup_logger
if TYPE_CHECKING:
from collections.abc import Callable
from flask import Flask
from werkzeug.wrappers import Response
from shelfmark.core.user_db import UserDB
logger = setup_logger(__name__)
STATS_PATH = "/api/stats"
def register_stats_routes(
app: Flask,
*,
get_user_db: Callable[[], UserDB | None],
resolve_auth_mode: Callable[[], str],
) -> None:
"""Register ``/api/stats``.
Both arguments are called per request so a runtime patch of either still takes effect.
"""
@app.route(STATS_PATH, methods=["GET"])
def api_stats() -> Response | tuple[Response, int]:
"""Counters for dashboards. Reachable with the read-only API key, an admin key or an admin session."""
scope = g.get("api_key_scope")
if resolve_auth_mode() != "none" and scope is None:
if "user_id" not in session:
return jsonify({"error": "Unauthorized"}), 401
if not session.get("is_admin", False):
return jsonify({"error": "Admin access required"}), 403
user_db = get_user_db()
if user_db is None:
return jsonify({"error": "User database unavailable"}), 503
try:
payload = stats.collect(user_db.db_path)
except sqlite3.Error:
logger.exception("stats: could not read the database")
return jsonify({"error": "Internal Server Error"}), 500
return jsonify(payload)
+7 -21
View File
@@ -94,6 +94,7 @@ from shelfmark.core.requests_service import (
sync_delivery_states_from_queue_status,
)
from shelfmark.core.socketio_origins import build_cors_allowed_origins
from shelfmark.core.stats_routes import STATS_PATH, register_stats_routes
from shelfmark.core.user_db import UserDB
from shelfmark.core.utils import AUDIOBOOK_FORMATS, normalize_base_path
from shelfmark.core.viewer_redaction import (
@@ -836,7 +837,7 @@ def _proxy_default_is_admin(db: UserDB) -> bool:
_API_KEY_EXEMPT_PREFIXES = ("/api/auth/",)
_API_KEY_EXEMPT_PATHS = frozenset({"/api/health"})
_READ_ONLY_KEY_PATHS = frozenset({"/api/stats"})
_READ_ONLY_KEY_PATHS = frozenset({STATS_PATH})
@app.before_request
@@ -1491,26 +1492,11 @@ def api_config() -> Response | tuple[Response, int]:
return jsonify({"error": "Could not load the configuration"}), 500
@app.route("/api/stats", methods=["GET"])
def api_stats() -> Response | tuple[Response, int]:
"""Counters for dashboards. Reachable with the read-only API key, an admin key or an admin session."""
from shelfmark.core import stats
scope = g.get("api_key_scope")
if get_auth_mode() != "none" and scope is None:
if "user_id" not in session:
return jsonify({"error": "Unauthorized"}), 401
if not session.get("is_admin", False):
return jsonify({"error": "Admin access required"}), 403
if user_db is None:
return jsonify({"error": "User database unavailable"}), 503
try:
payload = stats.collect(user_db.db_path)
except sqlite3.Error:
logger.exception("stats: could not read the database")
return jsonify({"error": "Internal Server Error"}), 500
return jsonify(payload)
register_stats_routes(
app,
get_user_db=lambda: user_db,
resolve_auth_mode=_resolve_auth_mode_for_routes,
)
@app.route("/api/health", methods=["GET"])