From 53c3250abefae12bd1f40da463e1ed03eea9829d Mon Sep 17 00:00:00 2001 From: Florian Date: Thu, 5 Nov 2026 10:46:15 +0100 Subject: [PATCH] Moved routes to their own modules --- src/main.py | 228 +-------------------------------------- src/routers/__init__.py | 0 src/routers/scheduler.py | 101 +++++++++++++++++ src/routers/user.py | 139 ++++++++++++++++++++++++ 4 files changed, 245 insertions(+), 223 deletions(-) create mode 100644 src/routers/__init__.py create mode 100644 src/routers/scheduler.py create mode 100644 src/routers/user.py diff --git a/src/main.py b/src/main.py index bb52053..d3b8fd6 100644 --- a/src/main.py +++ b/src/main.py @@ -3,23 +3,16 @@ import os from contextlib import asynccontextmanager import uvicorn -from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request -from requests.exceptions import RequestException +from fastapi import Depends, FastAPI 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 -from feed_handler import ( - build_gitea_runner, - ensure_feed_exists, - fetch_latest, -) +from db_module import close_connection_pool, create_connection_pool, start_healthcheck_thread from handlers import register_handlers from metrics_server import metrics_api -from models import FeedEntry, FeedEntryType, SubscribedFeed, SubscribedFeedList +from routers.scheduler import scheduler_router +from routers.user import user_router from secret_manager import cleanup_secret_files, get_secret -from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError -from service_auth import UserId +from send_notifications import NotificationClient from simple_logger_handler import LOG_LEVEL, setup_logger logger = setup_logger(__name__) @@ -58,217 +51,6 @@ api = FastAPI( ) register_handlers(api) -user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) -scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) - -@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 type = %s AND active = 1", - (rss_feed_type.value,), - ) - 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 rss_feeds: - results["checked"] += 1 - try: - 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 - - 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 online_information.feed_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={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={online_information.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, feed_name = %s WHERE id = %s", - (online_information.entry_id, online_information.feed_title, feed_id), - ) - db.commit() - logger.debug(f"[FastAPI] Notification sent and database updated for feed ID {feed_id}") - results["new"] += 1 - - 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: - db.rollback() - results["errors"] += 1 - logger.error( - f"[FastAPI] Failed for feed ID {feed_id} ({rss_feed}): {e}", - exc_info=True, - ) - - if results["errors"]: - logger.warning(f"[FastAPI] {rss_feed_type.value} check finished with errors: {results}") - raise HTTPException(status_code=500, detail=results) - - logger.info(f"[FastAPI] {rss_feed_type.value} check complete: {results}") - return results - -@user_router.post("/feed/add") -def add_feed( - user_id: UserId, - feed_entry: FeedEntry, - db = Depends(get_db) - ): - url = feed_entry.rss_feed_url - logger.info(f"[FastAPI] Subscribe request: RSS URL={url}, user_id={user_id}") - - with db.cursor() as cursor: - cursor.execute("SELECT id FROM rss_feeds WHERE rss_feed = %s", (url,),) - known = cursor.fetchone() - - last_entry_id = None - feed_name = None - - if known is None: - ensure_feed_exists(url) - logger.info(f"[FastApi] Adding new RSS feed, type={feed_entry.type}, rss_feed_url={url}") - try: - update = fetch_latest(feed_entry.type, url) - last_entry_id, feed_name = update.entry_id, update.feed_title - except (ValueError, RequestException) as e: - 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, feed_name, last_entry) VALUES (%s, %s, %s, %s) " - "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, feed_name = COALESCE(VALUES(feed_name), feed_name)", - (feed_entry.type.value, url, feed_name, last_entry_id,), - ) - feed_id = cursor.lastrowid - - cursor.execute ( - "INSERT INTO rss_feed_subscriptions (user_id, rss_feed_id, feed_name) " - "VALUES (%s, %s, %s) " - "ON DUPLICATE KEY UPDATE active = 1, feed_name = VALUES(feed_name)", - (user_id, feed_id, feed_entry.custom_name), - ) - db.commit() - except Exception as e: - db.rollback() - logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {url}") - raise HTTPException(status_code=500) from e - - logger.info(f"[FastAPI] user_id={user_id} subscribed to feed_id={feed_id}, url={url}") - return {"status":"ok","detail":"Subscribed to feed","feed_url":url} - -@user_router.post("/feed/delete") -def delete_feed( - user_id: UserId, - feed_entry: FeedEntry, - db = Depends(get_db), - ): - url = feed_entry.rss_feed_url - logger.info(f"[FastAPI] Unsubscribe request: RSS URL={url}, user_id={user_id}") - - try: - with db.cursor() as cursor: - cursor.execute( - "SELECT id FROM rss_feeds WHERE rss_feed = %s FOR UPDATE", - (url,), - ) - known = cursor.fetchone() - - if known is None: - db.rollback() - return {"status":"ok","detail":"Unsubscribed to feed","feed_url":url} - - feed_id = known[0] - cursor.execute( - "UPDATE rss_feed_subscriptions SET active = 0 " - "WHERE user_id = %s AND rss_feed_id = %s", - (user_id, feed_id), - ) - - cursor.execute( - "SELECT 1 from rss_feed_subscriptions " - "WHERE rss_feed_id = %s AND active = 1 LIMIT 1 FOR SHARE", - (feed_id,), - ) - if cursor.fetchone() is None: - cursor.execute( - "UPDATE rss_feeds SET active = 0 WHERE id = %s", - (feed_id,), - ) - - db.commit() - except Exception as e: - db.rollback() - logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {url}") - raise HTTPException(status_code=500) from e - return {"status":"ok","detail":"Unsubscribed to feed","feed_url":url} - -@user_router.get("/feed/list") -def list_subscribed_feeds(user_id: UserId, db=Depends(get_db)): - logger.info(f"[FastAPI] Received request to list feeds for User ID = {user_id}") - try: - with db.cursor() as cursor: - cursor.execute( - "SELECT s.feed_name as custom_name, r.feed_name as feed_name, r.type, r.rss_feed, s.active from rss_feed_subscriptions as s " - "JOIN rss_feeds as r on s.rss_feed_id = r.id " - "WHERE s.user_id = %s ORDER BY r.type", - (user_id,), - ) - rows = cursor.fetchall() - - feed_list = SubscribedFeedList() - for custom_name, feed_name, feed_type, rss_feed, is_active in rows: - entry = SubscribedFeed( - feed_name = custom_name or feed_name, - type = feed_type, - rss_feed = rss_feed, - ) - (feed_list.active if is_active else feed_list.inactive).append(entry) - except Exception as e: - logger.exception("[FastAPI] Failed to list feeds") - raise HTTPException(status_code=500) from e - return feed_list - api.include_router(user_router) api.include_router(scheduler_router) diff --git a/src/routers/__init__.py b/src/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/routers/scheduler.py b/src/routers/scheduler.py new file mode 100644 index 0000000..5a8b2b3 --- /dev/null +++ b/src/routers/scheduler.py @@ -0,0 +1,101 @@ + +from fastapi import APIRouter, Depends, HTTPException, Request + +from auth_config import Caller, auth +from db_module import get_db +from feed_handler import ( + build_gitea_runner, + fetch_latest, +) +from models import FeedEntryType +from send_notifications import NotificationError, NotificationUnavailableError +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) +scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) + +@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 type = %s AND active = 1", + (rss_feed_type.value,), + ) + 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 rss_feeds: + results["checked"] += 1 + try: + 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 + + 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 online_information.feed_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={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={online_information.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, feed_name = %s WHERE id = %s", + (online_information.entry_id, online_information.feed_title, feed_id), + ) + db.commit() + logger.debug(f"[FastAPI] Notification sent and database updated for feed ID {feed_id}") + results["new"] += 1 + + 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: + db.rollback() + results["errors"] += 1 + logger.error( + f"[FastAPI] Failed for feed ID {feed_id} ({rss_feed}): {e}", + exc_info=True, + ) + + if results["errors"]: + logger.warning(f"[FastAPI] {rss_feed_type.value} check finished with errors: {results}") + raise HTTPException(status_code=500, detail=results) + + logger.info(f"[FastAPI] {rss_feed_type.value} check complete: {results}") + return results \ No newline at end of file diff --git a/src/routers/user.py b/src/routers/user.py new file mode 100644 index 0000000..1c9586c --- /dev/null +++ b/src/routers/user.py @@ -0,0 +1,139 @@ + +from fastapi import APIRouter, Depends, HTTPException +from requests.exceptions import RequestException + +from auth_config import Caller, auth +from db_module import get_db +from feed_errors import FeedCheckFailedError +from feed_handler import ( + ensure_feed_exists, + fetch_latest, +) +from models import FeedEntry, SubscribedFeed, SubscribedFeedList +from service_auth import UserId +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) +user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) + +@user_router.post("/feed/add") +def add_feed( + user_id: UserId, + feed_entry: FeedEntry, + db = Depends(get_db) + ): + url = feed_entry.rss_feed_url + logger.info(f"[FastAPI] Subscribe request: RSS URL={url}, user_id={user_id}") + + with db.cursor() as cursor: + cursor.execute("SELECT id FROM rss_feeds WHERE rss_feed = %s", (url,),) + known = cursor.fetchone() + + last_entry_id = None + feed_name = None + + if known is None: + ensure_feed_exists(url) + logger.info(f"[FastApi] Adding new RSS feed, type={feed_entry.type}, rss_feed_url={url}") + try: + update = fetch_latest(feed_entry.type, url) + last_entry_id, feed_name = update.entry_id, update.feed_title + except (ValueError, RequestException) as e: + 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, feed_name, last_entry) VALUES (%s, %s, %s, %s) " + "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, feed_name = COALESCE(VALUES(feed_name), feed_name)", + (feed_entry.type.value, url, feed_name, last_entry_id,), + ) + feed_id = cursor.lastrowid + + cursor.execute ( + "INSERT INTO rss_feed_subscriptions (user_id, rss_feed_id, feed_name) " + "VALUES (%s, %s, %s) " + "ON DUPLICATE KEY UPDATE active = 1, feed_name = VALUES(feed_name)", + (user_id, feed_id, feed_entry.custom_name), + ) + db.commit() + except Exception as e: + db.rollback() + logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {url}") + raise HTTPException(status_code=500) from e + + logger.info(f"[FastAPI] user_id={user_id} subscribed to feed_id={feed_id}, url={url}") + return {"status":"ok","detail":"Subscribed to feed","feed_url":url} + +@user_router.post("/feed/delete") +def delete_feed( + user_id: UserId, + feed_entry: FeedEntry, + db = Depends(get_db), + ): + url = feed_entry.rss_feed_url + logger.info(f"[FastAPI] Unsubscribe request: RSS URL={url}, user_id={user_id}") + + try: + with db.cursor() as cursor: + cursor.execute( + "SELECT id FROM rss_feeds WHERE rss_feed = %s FOR UPDATE", + (url,), + ) + known = cursor.fetchone() + + if known is None: + db.rollback() + return {"status":"ok","detail":"Unsubscribed to feed","feed_url":url} + + feed_id = known[0] + cursor.execute( + "UPDATE rss_feed_subscriptions SET active = 0 " + "WHERE user_id = %s AND rss_feed_id = %s", + (user_id, feed_id), + ) + + cursor.execute( + "SELECT 1 from rss_feed_subscriptions " + "WHERE rss_feed_id = %s AND active = 1 LIMIT 1 FOR SHARE", + (feed_id,), + ) + if cursor.fetchone() is None: + cursor.execute( + "UPDATE rss_feeds SET active = 0 WHERE id = %s", + (feed_id,), + ) + + db.commit() + except Exception as e: + db.rollback() + logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {url}") + raise HTTPException(status_code=500) from e + return {"status":"ok","detail":"Unsubscribed to feed","feed_url":url} + +@user_router.get("/feed/list") +def list_subscribed_feeds(user_id: UserId, db=Depends(get_db)): + logger.info(f"[FastAPI] Received request to list feeds for User ID = {user_id}") + try: + with db.cursor() as cursor: + cursor.execute( + "SELECT s.feed_name as custom_name, r.feed_name as feed_name, r.type, r.rss_feed, s.active from rss_feed_subscriptions as s " + "JOIN rss_feeds as r on s.rss_feed_id = r.id " + "WHERE s.user_id = %s ORDER BY r.type", + (user_id,), + ) + rows = cursor.fetchall() + + feed_list = SubscribedFeedList() + for custom_name, feed_name, feed_type, rss_feed, is_active in rows: + entry = SubscribedFeed( + feed_name = custom_name or feed_name, + type = feed_type, + rss_feed = rss_feed, + ) + (feed_list.active if is_active else feed_list.inactive).append(entry) + except Exception as e: + logger.exception("[FastAPI] Failed to list feeds") + raise HTTPException(status_code=500) from e + return feed_list \ No newline at end of file