Moved routes to their own modules

This commit is contained in:
florian committed 2026-11-05 10:46:15 +01:00
1 parent 5980676452
commit 53c3250abe
4 files changed
+245 -223

No files matched your search

+5 -223
View File
@@ -3,23 +3,16 @@ import os
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
import uvicorn import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request from fastapi import Depends, FastAPI
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, start_healthcheck_thread
from feed_errors import FeedCheckFailedError
from feed_handler import (
build_gitea_runner,
ensure_feed_exists,
fetch_latest,
)
from handlers import register_handlers from handlers import register_handlers
from metrics_server import metrics_api 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 secret_manager import cleanup_secret_files, get_secret
from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from send_notifications import NotificationClient
from service_auth import UserId
from simple_logger_handler import LOG_LEVEL, setup_logger from simple_logger_handler import LOG_LEVEL, setup_logger
logger = setup_logger(__name__) logger = setup_logger(__name__)
@@ -58,217 +51,6 @@ api = FastAPI(
) )
register_handlers(api) 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(user_router)
api.include_router(scheduler_router) api.include_router(scheduler_router)
View File
Whitespace-only changes.
+101
View File
@@ -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
+139
View File
@@ -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