Merge pull request 'Feauture/unified path' (#8) from feauture/unified-path into main

Reviewed-on: #8
This commit was merged in pull request #8.
This commit is contained in:
florian committed 2026-11-03 12:06:20 +00:00
commit 5217d36636
5 files changed
+140 -198

No files matched your search

+18 -2
View File
@@ -1,35 +1,51 @@
class FeedError(Exception): class FeedError(Exception):
"""Base class for every feed problem the API can explain to a client.""" """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): 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. Subclasses ValueError so Pydantic reports it as a normal 422 validation error.
""" """
status_code = 422 status_code = 422
code = "srr_invalid_url"
client_visible = True
class WrongHostError(InvalidFeedUrlError): class WrongHostError(InvalidFeedUrlError):
code = "srr_wrong_host"
def __init__(self, feed_type: str, expected_host: str): def __init__(self, feed_type: str, expected_host: str):
super().__init__(f"A {feed_type} feed must be on {expected_host}") super().__init__(f"A {feed_type} feed must be on {expected_host}")
class UnsupportedPathError(InvalidFeedUrlError): class UnsupportedPathError(InvalidFeedUrlError):
code = "srr_unsupported_path"
def __init__(self, hint: str): def __init__(self, hint: str):
super().__init__(f"Unrecognised feed path ({hint})") super().__init__(f"Unrecognised feed path ({hint})")
class FeedNotFoundError(FeedError): class FeedNotFoundError(FeedError):
status_code = 422 status_code = 422
code = "srr_feed_not_found"
client_visible = True
def __init__(self, url: str): def __init__(self, url: str):
super().__init__(f"The feed does not exist: {url}") super().__init__(f"The feed does not exist: {url}")
class FeedCheckFailedError(FeedError): 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 status_code = 502
code = "srr_feed_check_failed"
def __init__(self, url: str): def __init__(self, url: str):
super().__init__(f"Could not verify the feed right now, try again later: {url}") super().__init__(f"Could not verify the feed right now, try again later: {url}")
+6 -6
View File
@@ -5,7 +5,7 @@ import feedparser
import requests import requests
from feed_errors import FeedCheckFailedError, FeedNotFoundError from feed_errors import FeedCheckFailedError, FeedNotFoundError
from models import FeedEntryType, FeedUpdate from models import FeedEntryType, GiteaUpdate, RoyalRoadUpdate
from simple_logger_handler import setup_logger from simple_logger_handler import setup_logger
logger = setup_logger(__name__) 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}") logger.debug(f"[Feed] Parsed {len(feed.entries)} entries from {rss_feed_url}")
return feed 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. 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. ValueError: If the feed has no entries.
""" """
feed = get_rss_feed_data(rss_feed_url) feed = get_rss_feed_data(rss_feed_url)
feed_update = FeedUpdate ( feed_update = GiteaUpdate (
entry_id = feed.entries[0]["link"], entry_id = feed.entries[0]["link"],
feed_title = feed.feed.title, feed_title = feed.feed.title,
body_title = feed.entries[0]["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}") logger.debug(f"[Feed] Latest release for feed '{rss_feed_url}': {feed_update.feed_title} -> {feed_update.link}")
return feed_update 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. 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"] chapter_title_with_story_title = latest_chapter_data["title"]
body_title = chapter_title_with_story_title.removeprefix(feed.feed.title).removeprefix(" - ") 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"]), entry_id = str(latest_chapter_data["id"]),
feed_title = feed.feed.title, feed_title = feed.feed.title,
body_title = body_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: if status != 200:
raise FeedCheckFailedError(url) 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: match feed_type:
case FeedEntryType.ROYALROAD: case FeedEntryType.ROYALROAD:
return grab_royal_road_chapter_information(url) return grab_royal_road_chapter_information(url)
+65
View File
@@ -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 feed_errors import FeedError
from metrics_server import REQUEST_COUNTER
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)
+43 -189
View File
@@ -1,25 +1,23 @@
import asyncio import asyncio
import os import os
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import assert_never
import uvicorn import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from requests.exceptions import RequestException from requests.exceptions import RequestException
from auth_config import Caller, auth from auth_config import Caller, auth
from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread 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 ( from feed_handler import (
build_gitea_runner, build_gitea_runner,
ensure_feed_exists, ensure_feed_exists,
fetch_latest, fetch_latest,
grab_gitea_release_information,
grab_royal_road_chapter_information,
) )
from metrics_server import REQUEST_COUNTER, metrics_api from handlers import register_handlers
from models import FeedEntry from metrics_server import metrics_api
from models import FeedEntry, FeedEntryType
from secret_manager import cleanup_secret_files, get_secret from secret_manager import cleanup_secret_files, get_secret
from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError
from service_auth import UserId from service_auth import UserId
@@ -60,89 +58,34 @@ api = FastAPI(
dependencies=[Depends(auth.authenticate)], dependencies=[Depends(auth.authenticate)],
) )
@api.exception_handler(FeedError) register_handlers(api)
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
user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
@scheduler_router.post("/gitea") @scheduler_router.post("/update/{rss_feed_type}")
def get_release_information(request: Request, db = Depends(get_db)): def update_rss_feeds(
""" request: Request,
Checks for new Gitea releases and updates the database accordingly. rss_feed_type: FeedEntryType,
db=Depends(get_db),
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")
results = {"checked": 0, "new": 0, "errors": 0} results = {"checked": 0, "new": 0, "errors": 0}
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"SELECT id, rss_feed, last_entry " "SELECT id, rss_feed, last_entry "
"FROM rss_feeds WHERE active = 1 AND type = %s", "FROM rss_feeds WHERE type = %s AND active = 1",
("gitea",), (rss_feed_type.value,),
) )
feeds = cursor.fetchall() rss_feeds = cursor.fetchall()
logger.debug(f"[FastAPI] Found {len(feeds)} active Gitea feeds to check.") 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 results["checked"] += 1
try: try:
gitea_release_information = grab_gitea_release_information(rss_feed) online_information = fetch_latest(rss_feed_type, rss_feed)
logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={gitea_release_information.entry_id}") 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:
if gitea_release_information.entry_id == last_entry:
continue continue
logger.info(f"[FastAPI] Release change for ID {feed_id}: {gitea_release_information.body_title}")
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"SELECT user_id, feed_name " "SELECT user_id, feed_name "
@@ -154,25 +97,31 @@ def get_release_information(request: Request, db = Depends(get_db)):
sent = 0 sent = 0
failed = 0 failed = 0
repo = gitea_release_information.link.split("/releases")[0].removeprefix("https://") match rss_feed_type:
for user_id,feed_name in subscribers: 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: try:
request.app.state.notifier.send( request.app.state.notifier.send(
title=feed_name or repo, title=feed_name or title,
body=gitea_release_information.body_title, body=online_information.body_title,
link=gitea_release_information.link, link=online_information.link,
category="utility", category=online_information.notification_type,
user_id=user_id, user_id=user_id,
) )
sent += 1 sent += 1
except NotificationUnavailableError: except NotificationUnavailableError:
failed = len(subscribers) - sent failed = len(subscribers) - sent
logger.error(f"[FastAPI] Notifications backend unavailable, skipping " logger.error (f"[FastAPI] Notifications backend unavailable, skipping "
f"{failed} remaining user(s) for feed_id={feed_id}") f"{failed} remaining user(s) for feed={online_information.feed_title}")
break break
except NotificationError as e: except NotificationError as e:
failed += 1 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)}") f"status={getattr(e, 'status_code',None)}")
if subscribers and sent == 0: if subscribers and sent == 0:
results["errors"] += 1 results["errors"] += 1
@@ -181,14 +130,14 @@ def get_release_information(request: Request, db = Depends(get_db)):
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"UPDATE rss_feeds SET last_entry = %s WHERE id = %s", "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() 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 results["new"] += 1
if gitea_release_information.link.startswith("https://gitea.com/gitea/runner/releases/tag/v"): 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(gitea_release_information.body_title, request.app.state.gitea_token) 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}") logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}")
except Exception as e: except Exception as e:
@@ -200,105 +149,10 @@ def get_release_information(request: Request, db = Depends(get_db)):
) )
if results["errors"]: 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) raise HTTPException(status_code=500, detail=results)
logger.info(f"[FastAPI] Gitea check complete: {results}") logger.info(f"[FastAPI] {rss_feed_type.value} 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}")
return results return results
@user_router.post("/add/feed") @user_router.post("/add/feed")
@@ -321,14 +175,14 @@ def add_feed(
try: try:
last_entry_id = fetch_latest(feed_entry.type, url).entry_id last_entry_id = fetch_latest(feed_entry.type, url).entry_id
except (ValueError, RequestException) as e: 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 raise FeedCheckFailedError(url) from e
try: try:
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"INSERT into rss_feeds (type,rss_feed,last_entry) VALUES (%s,%s,%s) " "INSERT into rss_feeds (type,rss_feed,last_entry) VALUES (%s,%s,%s) "
"ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1", "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 feed_id = cursor.lastrowid
+8 -1
View File
@@ -1,6 +1,6 @@
from dataclasses import dataclass from dataclasses import dataclass
from enum import StrEnum from enum import StrEnum
from typing import Self from typing import ClassVar, Self
from urllib.parse import urlsplit from urllib.parse import urlsplit
from pydantic import BaseModel, Field, field_validator, model_validator from pydantic import BaseModel, Field, field_validator, model_validator
@@ -14,12 +14,19 @@ class FeedUpdate():
entry_id: str entry_id: str
feed_title: str feed_title: str
body_title: str body_title: str
notification_type: ClassVar[str]
link: str | None = None link: str | None = None
class FeedEntryType(StrEnum): class FeedEntryType(StrEnum):
ROYALROAD = "royalroad" ROYALROAD = "royalroad"
GITEA = "gitea" 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]: def _path_segments(raw: str, feed_type: FeedEntryType, host: str) -> list[str]:
"""Check scheme and host, then return the non-empty path segments.""" """Check scheme and host, then return the non-empty path segments."""