From 00b1da5321aa581980ef56f9cddeabfca9deaf81 Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 3 Nov 2026 11:45:55 +0100 Subject: [PATCH 1/4] Moved handlers to its own module, updated errors so they can propagate upstream if necessary --- src/feed_errors.py | 20 ++++++++++++-- src/handlers.py | 65 ++++++++++++++++++++++++++++++++++++++++++++++ src/main.py | 50 +++-------------------------------- 3 files changed, 87 insertions(+), 48 deletions(-) create mode 100644 src/handlers.py diff --git a/src/feed_errors.py b/src/feed_errors.py index fdb4f9a..16f4c69 100644 --- a/src/feed_errors.py +++ b/src/feed_errors.py @@ -1,35 +1,51 @@ class FeedError(Exception): """Base class for every feed problem the API can explain to a client.""" - status_code = 400 + status_code = 502 + code = "srr_upstream_error" + client_visible = False + + @property + def client_message(self) -> str: + return str(self) if self.client_visible else "The feed could not be checked right now, please try again later" class InvalidFeedUrlError(FeedError, ValueError): - """The URL is malformed or doesn't fit the chosen feed type. + """The URL is malformed or doesn't fit the chosen feed type. Raise it with the message to show. Subclasses ValueError so Pydantic reports it as a normal 422 validation error. """ status_code = 422 + code = "srr_invalid_url" + client_visible = True class WrongHostError(InvalidFeedUrlError): + code = "srr_wrong_host" + def __init__(self, feed_type: str, expected_host: str): super().__init__(f"A {feed_type} feed must be on {expected_host}") class UnsupportedPathError(InvalidFeedUrlError): + code = "srr_unsupported_path" + def __init__(self, hint: str): super().__init__(f"Unrecognised feed path ({hint})") class FeedNotFoundError(FeedError): status_code = 422 + code = "srr_feed_not_found" + client_visible = True def __init__(self, url: str): super().__init__(f"The feed does not exist: {url}") class FeedCheckFailedError(FeedError): + """The feed exists as far as we know, but checking it failed. The message names the URL, so it stays hidden.""" status_code = 502 + code = "srr_feed_check_failed" def __init__(self, url: str): super().__init__(f"Could not verify the feed right now, try again later: {url}") \ No newline at end of file diff --git a/src/handlers.py b/src/handlers.py new file mode 100644 index 0000000..89e8625 --- /dev/null +++ b/src/handlers.py @@ -0,0 +1,65 @@ +from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from starlette.middleware.base import BaseHTTPMiddleware + +from metrics_server import REQUEST_COUNTER +from feed_errors import FeedError +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + + +async def feed_error_handler(request: Request, exc: FeedError): + log = logger.warning if exc.client_visible else logger.error + log(f"[FastAPI] {request.method} {request.url.path} failed: {exc}") + return JSONResponse( + status_code=exc.status_code, + content={"code": exc.code, "detail": exc.client_message}, + ) + + +async def validation_error_handler(request: Request, exc: RequestValidationError): + errors = exc.errors() + + for err in errors: + original = (err.get("ctx") or {}).get("error") + if isinstance(original, FeedError): + return JSONResponse( + status_code=original.status_code, + content={"code": original.code, "detail": original.client_message}, + ) + + # Problems in headers, path or query are the calling service's fault, not the user's + if any(err["loc"][0] != "body" for err in errors): + logger.error(f"[FastAPI] Invalid request from a calling service: {[(e['loc'], e['msg']) for e in errors]}") + return JSONResponse(status_code=400, content={"code": "bad_request", "detail": "The request could not be processed"}) + + messages = [] + for err in errors: + field = ".".join(str(p) for p in err["loc"] if p != "body") + msg = err["msg"].removeprefix("Value error, ") + messages.append(f"{field}: {msg}" if field else msg) + return JSONResponse(status_code=422, content={"code": "validation_error", "detail": "; ".join(messages)}) + + +async def prometheus_middleware(request: Request, call_next): + logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}") + status = 500 + try: + response = await call_next(request) + status = response.status_code + logger.debug(f"[FastAPI] Request processed with status {status}") + except Exception as e: + logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True) + raise + finally: + REQUEST_COUNTER.labels(request.method, request.url.path, status).inc() + logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]") + return response + +def register_handlers(app: FastAPI) -> None: + """Attach the error handlers and the metrics middleware to the app.""" + app.add_exception_handler(FeedError, feed_error_handler) + app.add_exception_handler(RequestValidationError, validation_error_handler) + app.add_middleware(BaseHTTPMiddleware, dispatch=prometheus_middleware) diff --git a/src/main.py b/src/main.py index 6a8d271..2bd458b 100644 --- a/src/main.py +++ b/src/main.py @@ -4,13 +4,11 @@ from contextlib import asynccontextmanager import uvicorn from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request -from fastapi.exceptions import RequestValidationError -from fastapi.responses import JSONResponse from requests.exceptions import RequestException from auth_config import Caller, auth from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread -from feed_errors import FeedCheckFailedError, FeedError +from feed_errors import FeedCheckFailedError from feed_handler import ( build_gitea_runner, ensure_feed_exists, @@ -18,7 +16,8 @@ from feed_handler import ( grab_gitea_release_information, grab_royal_road_chapter_information, ) -from metrics_server import REQUEST_COUNTER, metrics_api +from handlers import register_handlers +from metrics_server import metrics_api from models import FeedEntry from secret_manager import cleanup_secret_files, get_secret from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError @@ -60,48 +59,7 @@ api = FastAPI( dependencies=[Depends(auth.authenticate)], ) -@api.exception_handler(FeedError) -async def feed_error_handler(request: Request, exc: FeedError): - return JSONResponse(status_code=exc.status_code, content={"detail":str(exc)}) - -@api.exception_handler(RequestValidationError) -async def validation_error_handler(request: Request, exc: RequestValidationError): - errors = exc.errors() - - # 1) If one of our own FeedErrors caused it, use its message as is - for err in errors: - original = (err.get("ctx") or {}).get("error") - if isinstance(original, FeedError): - return JSONResponse( - status_code=original.status_code, - content={"detail": str(original)}, - ) - - # 2) Anything else (missing field, bad enum value, bad header): flatten to one string - messages = [] - for err in errors: - field = ".".join(str(p) for p in err["loc"] if p != "body") - msg = err["msg"].removeprefix("Value error, ") - messages.append(f"{field}: {msg}" if field else msg) - return JSONResponse(status_code=422, content={"detail": "; ".join(messages)}) - -@api.middleware("http") -async def prometheus_middleware(request: Request, call_next): - logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}") - status = 500 - try: - response = await call_next(request) - status = response.status_code - logger.debug(f"[FastAPI] Request processed with status {status}") - except Exception as e: - logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True) - raise - finally: - REQUEST_COUNTER.labels(request.method, request.url.path, status).inc() - logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]") - return response - - +register_handlers(api) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) From 55fad17310c03ef5b26a8ca2554517edf2839c39 Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 3 Nov 2026 12:50:45 +0100 Subject: [PATCH 2/4] Linting --- src/handlers.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/handlers.py b/src/handlers.py index 89e8625..3a6f9f1 100644 --- a/src/handlers.py +++ b/src/handlers.py @@ -3,8 +3,8 @@ from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware -from metrics_server import REQUEST_COUNTER from feed_errors import FeedError +from metrics_server import REQUEST_COUNTER from simple_logger_handler import setup_logger logger = setup_logger(__name__) From f5b58170ecb78e4c548ffcbd3d1005e61e55ef4c Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 3 Nov 2026 13:01:48 +0100 Subject: [PATCH 3/4] Added subclasses that denote which notification type should be sent --- src/feed_handler.py | 12 ++++++------ src/models.py | 9 ++++++++- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/src/feed_handler.py b/src/feed_handler.py index 645c5a0..97a58d5 100644 --- a/src/feed_handler.py +++ b/src/feed_handler.py @@ -5,7 +5,7 @@ import feedparser import requests from feed_errors import FeedCheckFailedError, FeedNotFoundError -from models import FeedEntryType, FeedUpdate +from models import FeedEntryType, GiteaUpdate, RoyalRoadUpdate from simple_logger_handler import setup_logger logger = setup_logger(__name__) @@ -54,7 +54,7 @@ def get_rss_feed_data(rss_feed_url: str, max_retries: int = 3) -> feedparser.Fee logger.debug(f"[Feed] Parsed {len(feed.entries)} entries from {rss_feed_url}") return feed -def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate: +def grab_gitea_release_information(rss_feed_url: str) -> GiteaUpdate: """ Fetches the latest release information from a Gitea releases RSS feed. @@ -65,7 +65,7 @@ def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate: ValueError: If the feed has no entries. """ feed = get_rss_feed_data(rss_feed_url) - feed_update = FeedUpdate ( + feed_update = GiteaUpdate ( entry_id = feed.entries[0]["link"], feed_title = feed.feed.title, body_title = feed.entries[0]["title"], @@ -75,7 +75,7 @@ def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate: logger.debug(f"[Feed] Latest release for feed '{rss_feed_url}': {feed_update.feed_title} -> {feed_update.link}") return feed_update -def grab_royal_road_chapter_information(rss_feed_url: str) -> FeedUpdate : +def grab_royal_road_chapter_information(rss_feed_url: str) -> RoyalRoadUpdate : """ Fetches the latest chapter information from a Royalroad RSS feed. @@ -93,7 +93,7 @@ def grab_royal_road_chapter_information(rss_feed_url: str) -> FeedUpdate : chapter_title_with_story_title = latest_chapter_data["title"] body_title = chapter_title_with_story_title.removeprefix(feed.feed.title).removeprefix(" - ") - feed_update = FeedUpdate ( + feed_update = RoyalRoadUpdate ( entry_id = str(latest_chapter_data["id"]), feed_title = feed.feed.title, body_title = body_title, @@ -143,7 +143,7 @@ def ensure_feed_exists(url: str, timeout: tuple[float, float] = (3.0, 5.0)) -> N if status != 200: raise FeedCheckFailedError(url) -def fetch_latest(feed_type: FeedEntryType, url: str) -> FeedUpdate: +def fetch_latest(feed_type: FeedEntryType, url: str) -> RoyalRoadUpdate | GiteaUpdate: match feed_type: case FeedEntryType.ROYALROAD: return grab_royal_road_chapter_information(url) diff --git a/src/models.py b/src/models.py index 444c05e..35409bc 100644 --- a/src/models.py +++ b/src/models.py @@ -1,6 +1,6 @@ from dataclasses import dataclass from enum import StrEnum -from typing import Self +from typing import ClassVar, Self from urllib.parse import urlsplit from pydantic import BaseModel, Field, field_validator, model_validator @@ -14,12 +14,19 @@ class FeedUpdate(): entry_id: str feed_title: str body_title: str + notification_type: ClassVar[str] link: str | None = None class FeedEntryType(StrEnum): ROYALROAD = "royalroad" GITEA = "gitea" +class RoyalRoadUpdate(FeedUpdate): + notification_type = "royal-road" + + +class GiteaUpdate(FeedUpdate): + notification_type = "utility" def _path_segments(raw: str, feed_type: FeedEntryType, host: str) -> list[str]: """Check scheme and host, then return the non-empty path segments.""" From 1b451620a72c47f99ea5a1724d51ba4c5bfe660d Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 3 Nov 2026 13:03:38 +0100 Subject: [PATCH 4/4] Added "/update/{rss_feed_type}" path that handles both royalroad and gitea --- src/main.py | 182 +++++++++++----------------------------------------- 1 file changed, 39 insertions(+), 143 deletions(-) diff --git a/src/main.py b/src/main.py index 2bd458b..d830ee9 100644 --- a/src/main.py +++ b/src/main.py @@ -1,6 +1,7 @@ import asyncio import os from contextlib import asynccontextmanager +from typing import assert_never import uvicorn from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request @@ -13,12 +14,10 @@ from feed_handler import ( build_gitea_runner, ensure_feed_exists, fetch_latest, - grab_gitea_release_information, - grab_royal_road_chapter_information, ) from handlers import register_handlers from metrics_server import metrics_api -from models import FeedEntry +from models import FeedEntry, FeedEntryType from secret_manager import cleanup_secret_files, get_secret from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from service_auth import UserId @@ -63,44 +62,30 @@ register_handlers(api) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) -@scheduler_router.post("/gitea") -def get_release_information(request: Request, db = Depends(get_db)): - """ - Checks for new Gitea releases and updates the database accordingly. - - Behaviour: - - Fetches all active Gitea feeds and their last known release link. - - Queries each feed for its latest release. - - Updates `last_entry` if the latest release link differs from the stored one. - - Sends a notification for each changed release. - - A failure on one feed is logged and the remaining feeds are still checked. - - Returns: - A JSON object with checked/new/error counts. If any feed failed, - a 500 is returned with the same counts as the detail. - """ - logger.info("[FastAPI] Checking for new Gitea releases") +@scheduler_router.post("/update/{rss_feed_type}") +def update_rss_feeds( + request: Request, + rss_feed_type: FeedEntryType, + db=Depends(get_db), + ): results = {"checked": 0, "new": 0, "errors": 0} with db.cursor() as cursor: cursor.execute( "SELECT id, rss_feed, last_entry " - "FROM rss_feeds WHERE active = 1 AND type = %s", - ("gitea",), + "FROM rss_feeds WHERE type = %s AND active = 1", + (rss_feed_type.value,), ) - feeds = cursor.fetchall() - logger.debug(f"[FastAPI] Found {len(feeds)} active Gitea feeds to check.") + rss_feeds = cursor.fetchall() + logger.debug(f"[FastAPI] Found {len(rss_feeds)} active {rss_feed_type.value} RSS feeds to check.") - for feed_id, rss_feed, last_entry in feeds: + for feed_id, rss_feed, last_entry in rss_feeds: results["checked"] += 1 try: - gitea_release_information = grab_gitea_release_information(rss_feed) - logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={gitea_release_information.entry_id}") - - if gitea_release_information.entry_id == last_entry: + online_information = fetch_latest(rss_feed_type, rss_feed) + logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={online_information.entry_id}") + if online_information.entry_id == last_entry: continue - logger.info(f"[FastAPI] Release change for ID {feed_id}: {gitea_release_information.body_title}") - with db.cursor() as cursor: cursor.execute( "SELECT user_id, feed_name " @@ -112,25 +97,31 @@ def get_release_information(request: Request, db = Depends(get_db)): sent = 0 failed = 0 - repo = gitea_release_information.link.split("/releases")[0].removeprefix("https://") - for user_id,feed_name in subscribers: + match rss_feed_type: + case FeedEntryType.GITEA: + title = online_information.link.split("/releases")[0].removeprefix("https://") + case FeedEntryType.ROYALROAD: + title = online_information.feed_title + case _: + assert_never(rss_feed_type) + for user_id, feed_name in subscribers: try: request.app.state.notifier.send( - title=feed_name or repo, - body=gitea_release_information.body_title, - link=gitea_release_information.link, - category="utility", + title=feed_name or title, + body=online_information.body_title, + link=online_information.link, + category=online_information.notification_type, user_id=user_id, ) sent += 1 except NotificationUnavailableError: failed = len(subscribers) - sent - logger.error(f"[FastAPI] Notifications backend unavailable, skipping " - f"{failed} remaining user(s) for feed_id={feed_id}") + logger.error (f"[FastAPI] Notifications backend unavailable, skipping " + f"{failed} remaining user(s) for feed={online_information.feed_title}") break except NotificationError as e: failed += 1 - logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for feed_id={feed_id} " + logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for feed={online_information.feed_title} " f"status={getattr(e, 'status_code',None)}") if subscribers and sent == 0: results["errors"] += 1 @@ -139,14 +130,14 @@ def get_release_information(request: Request, db = Depends(get_db)): with db.cursor() as cursor: cursor.execute( "UPDATE rss_feeds SET last_entry = %s WHERE id = %s", - (gitea_release_information.entry_id, feed_id), + (online_information.entry_id, feed_id), ) db.commit() - logger.debug(f"[FastAPI] Notification sent and database updated for Gitea feed ID {feed_id}") + logger.debug(f"[FastAPI] Notification sent and database updated for feed ID {feed_id}") results["new"] += 1 - if gitea_release_information.link.startswith("https://gitea.com/gitea/runner/releases/tag/v"): - gitea_build_success = build_gitea_runner(gitea_release_information.body_title, request.app.state.gitea_token) + if rss_feed_type is FeedEntryType.GITEA and (online_information.link or "").startswith("https://gitea.com/gitea/runner/releases/tag/v"): + gitea_build_success = build_gitea_runner(online_information.body_title, request.app.state.gitea_token) logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}") except Exception as e: @@ -158,105 +149,10 @@ def get_release_information(request: Request, db = Depends(get_db)): ) if results["errors"]: - logger.warning(f"[FastAPI] Gitea check finished with errors: {results}") + logger.warning(f"[FastAPI] {rss_feed_type.value} check finished with errors: {results}") raise HTTPException(status_code=500, detail=results) - logger.info(f"[FastAPI] Gitea check complete: {results}") - return results - -@scheduler_router.post("/royalroad") -def get_chapters(request: Request, db = Depends(get_db)): - """ - Checks for new Royalroad chapters and updates the database accordingly. - - Behaviour: - - Fetches all active stories and their last known chapter ID. - - Queries Royalroad for each story's latest chapter. - - Updates `last_entry` if a new chapter is found. - - Sends a notification for any newly discovered chapters. - - A failure on one story is logged and the remaining stories are still checked. - - Returns: - A JSON object with checked/new/error counts. If any story failed, - a 500 is returned with the same counts as the detail. - """ - logger.info("[FastAPI] Checking for new Royalroad chapters") - results = {"checked": 0, "new": 0, "errors": 0} - with db.cursor() as cursor: - cursor.execute( - "SELECT id, rss_feed, last_entry " - "FROM rss_feeds WHERE active = 1 AND type = %s", - ("royalroad",), - ) - feeds = cursor.fetchall() - logger.debug(f"[FastAPI] Found {len(feeds)} active stories to check.") - - for feed_id, rss_feed, last_entry in feeds: - results["checked"] += 1 - try: - story = grab_royal_road_chapter_information(rss_feed) - logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={story.entry_id}") - - if story.entry_id == last_entry: - continue - - logger.info(f"[FastAPI] New chapter detected for story ID {feed_id}: {story.feed_title}") - with db.cursor() as cursor: - cursor.execute ( - "SELECT user_id, feed_name " - "FROM rss_feed_subscriptions " - "WHERE rss_feed_id = %s AND active = 1", - (feed_id,), - ) - subscribers = cursor.fetchall() - - sent = 0 - failed = 0 - for user_id, feed_name in subscribers: - try: - request.app.state.notifier.send( - title=feed_name or story.feed_title, - body=story.body_title, - link=story.link, - category="royal-road", - user_id=user_id, - ) - sent += 1 - except NotificationUnavailableError: - failed = len(subscribers) - sent - logger.error (f"[FastAPI] Notifications backend unavailable, skipping " - f"{failed} remaining user(s) for story={story.feed_title}") - break - except NotificationError as e: - failed += 1 - logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for story={story.feed_title} " - f"status={getattr(e, 'status_code',None)}") - if subscribers and sent == 0: - results["errors"] += 1 - logger.error(f"[FastAPI] All notifications failed for Feed ID:{feed_id}, will retry later") - continue - with db.cursor() as cursor: - cursor.execute( - "UPDATE rss_feeds SET last_entry = %s WHERE id = %s", - (story.entry_id, feed_id), - ) - db.commit() - logger.debug(f"[FastAPI] Notification sent and database updated for story ID {feed_id}") - results["new"] += 1 - - except Exception as e: - db.rollback() - results["errors"] += 1 - logger.error( - f"[Royalroad] Failed for feed ID {feed_id} ({rss_feed}): {e}", - exc_info=True, - ) - - if results["errors"]: - logger.warning(f"[FastAPI] Royalroad check finished with errors: {results}") - raise HTTPException(status_code=500, detail=results) - - logger.info(f"[FastAPI] Royalroad check complete: {results}") + logger.info(f"[FastAPI] {rss_feed_type.value} check complete: {results}") return results @user_router.post("/add/feed") @@ -279,14 +175,14 @@ def add_feed( try: last_entry_id = fetch_latest(feed_entry.type, url).entry_id except (ValueError, RequestException) as e: - logger.warning(f"[FastAPI] Could not read latest entry for {url}, storing none") + logger.warning(f"[FastAPI] Could not read latest entry for {url}") raise FeedCheckFailedError(url) from e try: with db.cursor() as cursor: cursor.execute( "INSERT into rss_feeds (type,rss_feed,last_entry) VALUES (%s,%s,%s) " "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1", - (feed_entry.type, url, last_entry_id), + (feed_entry.type.value, url, last_entry_id), ) feed_id = cursor.lastrowid