From cd0278d8a8a9ab7f0b062e62841368f2f13304d0 Mon Sep 17 00:00:00 2001 From: Florian Date: Thu, 5 Nov 2026 12:45:14 +0100 Subject: [PATCH] Moved routes to their own modules --- src/main.py | 299 +-------------------------------------- src/routers/__init__.py | 0 src/routers/admin.py | 42 ++++++ src/routers/scheduler.py | 108 ++++++++++++++ src/routers/user.py | 165 +++++++++++++++++++++ src/seadex_client.py | 14 ++ 6 files changed, 333 insertions(+), 295 deletions(-) create mode 100644 src/routers/__init__.py create mode 100644 src/routers/admin.py create mode 100644 src/routers/scheduler.py create mode 100644 src/routers/user.py create mode 100644 src/seadex_client.py diff --git a/src/main.py b/src/main.py index b563f73..605a78f 100644 --- a/src/main.py +++ b/src/main.py @@ -1,32 +1,26 @@ import asyncio import os from contextlib import asynccontextmanager -from datetime import UTC, datetime import uvicorn -from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request -from seadex import EntryNotFoundError, SeaDexEntry -from service_auth import UserId +from fastapi import Depends, FastAPI -from anilist_query import get_anime_name from auth_config import Caller, auth from db_module import ( close_connection_pool, create_connection_pool, - get_db, start_healthcheck_thread, ) from handlers import register_handlers from metrics_server import metrics_api -from models import AnimeCreate, AnimeEntry, AnimeList, AnimeRemove +from routers.scheduler import scheduler_router +from routers.user import admin_router, user_router from secret_manager import cleanup_secret_files -from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError +from send_notifications import NotificationClient from simple_logger_handler import LOG_LEVEL, setup_logger logger = setup_logger(__name__) - - @asynccontextmanager async def lifespan(app: FastAPI): logger.info("[App] Starting application...") @@ -60,291 +54,6 @@ api = FastAPI( ) register_handlers(api) -user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) -scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) -admin_router = APIRouter(dependencies=[Depends(auth.require(Caller.ADMIN))]) - -def to_iso(dt: datetime) -> str: - return dt.astimezone(UTC).isoformat(timespec="microseconds") - - -def check_for_update(seadex: SeaDexEntry, anilist_id: int) -> str | None: - try: - return to_iso(seadex.from_id(anilist_id).updated_at) - except EntryNotFoundError: - return None - - -@scheduler_router.post("/update") -def check_seadex(request:Request, db=Depends(get_db)): - results = {"checked": 0, "new": 0, "errors": 0} - - with db.cursor() as cursor: - cursor.execute( - "SELECT id, anilist_id, anime_name, last_updated " - "FROM seadex_state WHERE active = 1" - ) - anime = cursor.fetchall() - - logger.debug("[FastAPI] Found %s active anime to check.", len(anime)) - if not anime: - logger.info("[FastAPI] No active Seadex entries") - return results - - with SeaDexEntry() as seadex: - for seadex_state_id, anilist_id, anime_name, last_updated in anime: - results["checked"] += 1 - try: - current = check_for_update(seadex, anilist_id) - if current is None: - logger.debug("[FastAPI] No entry yet for Anime: %s on SeaDex", anime_name) - continue - if current == last_updated: - continue - - logger.info( - "[FastAPI] Anime: %s updated, Anilist ID: %s", - anime_name,anilist_id, - ) - with db.cursor() as cursor: - cursor.execute( - "SELECT user_id, custom_name " - "FROM seadex_state_subscriptions " - "WHERE seadex_state_id = %s AND active = 1", - (seadex_state_id,), - ) - subscribers = cursor.fetchall() - - sent = 0 - failed = 0 - for user_id, custom_name in subscribers: - try: - request.app.state.notifier.send( - title=custom_name or anime_name, - body="SeaDex entry updated", - link=f"https://seadex.moe/{anilist_id}/", - user_id=user_id, - ) - sent += 1 - except NotificationUnavailableError: - failed = len(subscribers) - sent - logger.error("[FastAPI] Notifications backend unavailable, skipping " - "%s remaining user(s) for Anime: %s", - failed, anime_name, - ) - break - except NotificationError as e: - failed += 1 - logger.error( - "[FastAPI] Failed to send notification to user_id=%s for Anime: %s, status=%s", - user_id, anime_name, getattr(e, 'status_code',None), - ) - if subscribers and sent == 0: - results["errors"] += 1 - logger.error("[FastAPI] All notifications failed for Anime: %s, will retry later", anime_name) - continue - - with db.cursor() as cursor: - cursor.execute( - "UPDATE seadex_state SET last_updated = %s WHERE id = %s", - (current, seadex_state_id), - ) - db.commit() - logger.debug("[FastAPI] Notification sent and database updated for Anime: %s", anime_name) - results["new"] += 1 - - except Exception: - db.rollback() - results["errors"] += 1 - logger.exception( - "[FastAPI] Failed for id: %s Anilist: %s", - seadex_state_id, anilist_id, - ) - - if results["errors"]: - logger.warning("[FastAPI] SeaDex check finished with errors: %s", results) - raise HTTPException(status_code=500, detail=results) - - logger.info("[FastAPI] SeaDex check complete: %s", results) - return results - -@user_router.post("/anime/add") -def add_anime( - anime: AnimeCreate, - user_id: UserId, - db=Depends(get_db), - ): - logger.info("[FastAPI] Received request to track anilist_id=%s, user_id=%s", anime.anilist_id, user_id) - - with db.cursor() as cursor: - cursor.execute( - "SELECT id, active FROM seadex_state WHERE anilist_id = %s", - (anime.anilist_id,), - ) - known = cursor.fetchone() - - last_updated = None - with SeaDexEntry() as seadex: - if known is None: - anime_name = get_anime_name(anime.anilist_id) - last_updated = check_for_update(seadex, anime.anilist_id) - else: - anime_name, active = known - if not active: - last_updated = check_for_update(seadex, anime.anilist_id) - - with db.cursor() as cursor: - try: - cursor.execute( - "INSERT INTO seadex_state (anilist_id, anime_name, last_updated) " - "VALUES (%s, %s, %s) " - "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, " - "last_updated = COALESCE(VALUES(last_updated), last_updated)", - (anime.anilist_id, anime_name, last_updated), - ) - seadex_state_id = cursor.lastrowid - - cursor.execute( - "INSERT INTO seadex_state_subscriptions (user_id, seadex_state_id, custom_name) " - "VALUES (%s, %s, %s) " - "ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)", - (user_id, seadex_state_id, anime.custom_name or None), - ) - db.commit() - except Exception as e: - db.rollback() - logger.exception( - "[FastAPI] Failed for anilist_id=%s, name=%s, user_id=%s", - anime.anilist_id, anime_name, user_id, - ) - raise HTTPException(status_code=500) from e - - logger.info( - "[FastAPI] Tracking anilist_id=%s name=%r for user_id=%s (state_id=%s)", - anime.anilist_id, anime_name, user_id, seadex_state_id, - ) - return {"status": "ok", "detail": f"Tracking name={anime.custom_name or anime_name}, Anilist ID={anime.anilist_id}"} - -@user_router.post("/anime/delete") -def remove_anime( - anime: AnimeRemove, - user_id: UserId, - db=Depends(get_db) -): - logger.info( - "[FastAPI] Received request to deactivate anilist_id=%s for user_id=%s", - anime.anilist_id, user_id, - ) - with db.cursor() as cursor: - try: - cursor.execute( - "SELECT id FROM seadex_state " - "WHERE anilist_id = %s FOR UPDATE", - (anime.anilist_id,), - ) - known = cursor.fetchone() - - if known is None: - db.rollback() - logger.info("[FastAPI] Anime anilist_id=%s is not tracked, nothing to do", anime.anilist_id) - return {"status": "ok", "detail": f"Stopped tracking Anilist ID={anime.anilist_id}"} - - seadex_state_id = known[0] - cursor.execute( - "UPDATE seadex_state_subscriptions SET active = 0 " - "WHERE user_id = %s AND seadex_state_id = %s", - (user_id, seadex_state_id), - ) - - cursor.execute( - "SELECT 1 from seadex_state_subscriptions " - "WHERE seadex_state_id = %s AND active = 1 LIMIT 1", - (seadex_state_id,), - ) - - if cursor.fetchone() is None: - cursor.execute( - "UPDATE seadex_state SET active = 0 WHERE id = %s", - (seadex_state_id,), - ) - logger.info("[FastAPI] No active subscribers left, deactivated state_id=%s", seadex_state_id) - - db.commit() - except Exception as e: - db.rollback() - logger.exception( - "[FastAPI] Failed to unsubscribe anime anilist_id=%s for user_id=%s", - anime.anilist_id, user_id, - ) - raise HTTPException(status_code=500) from e - - logger.info( - "[FastAPI] Successfully deactivated tracking for anilist_id=%s for user_id=%s", - anime.anilist_id, user_id, - ) - return {"status": "ok", "detail": f"Stopped tracking Anilist ID={anime.anilist_id}"} - -@user_router.get("/anime/list", response_model=AnimeList) -def list_active_and_inactive_anime( - user_id: UserId, - db=Depends(get_db), - ): - logger.info("[FastAPI] Received request to list anime for User ID = %s", user_id) - - try: - with db.cursor() as cursor: - cursor.execute( - "SELECT s.custom_name AS custom_name, st.anime_name AS anime_name, " - "st.anilist_id, s.active, st.last_updated from seadex_state_subscriptions AS s " - "JOIN seadex_state AS st on s.seadex_state_id = st.id " - "WHERE s.user_id = %s ORDER BY st.anilist_id", - (user_id,), - ) - rows = cursor.fetchall() - - anime_list=AnimeList() - for custom_name, anime_name, anilist_id, is_active, last_updated in rows: - entry = AnimeEntry( - name=custom_name or anime_name, - anilist_id=anilist_id, - last_updated=None if last_updated is None else last_updated, - ) - (anime_list.active if is_active else anime_list.inactive).append(entry) - except Exception as e: - logger.exception("[FastAPI] Failed to list anime") - raise HTTPException(status_code=500) from e - - return anime_list - -@admin_router.post("/admin/update/names") -def update_all_names_to_most_recently_available(db=Depends(get_db)): - logger.info("[FastAPI] ADMIN: Got request to update all existing names") - try: - with db.cursor() as cursor: - cursor.execute( - "SELECT id, anilist_id from seadex_state" - ) - rows = cursor.fetchall() - for id, anilist_id in rows: - anime_name = get_anime_name(anilist_id) - cursor.execute( - "UPDATE seadex_state SET anime_name = %s " - "WHERE id = %s", - (anime_name,id), - ) - db.commit() - logger.info( - "[FastAPI] ADMIN: Updated id=%s, anilist_id=%s with name=%s", - id, anilist_id, anime_name, - ) - except Exception as e: - db.rollback() - logger.exception("[FastAPI] ADMIN: Failed to update all existing names") - raise HTTPException(status_code=500) from e - - logger.info("[FastAPI] ADMIN: Updated %s entries", len(rows)) - return {"status": "ok", "detail": f"Updated {len(rows)} entries"} - api.include_router(user_router) api.include_router(scheduler_router) api.include_router(admin_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/admin.py b/src/routers/admin.py new file mode 100644 index 0000000..8a1a3c6 --- /dev/null +++ b/src/routers/admin.py @@ -0,0 +1,42 @@ + +from fastapi import APIRouter, Depends, HTTPException + +from anilist_query import get_anime_name +from auth_config import Caller, auth +from db_module import ( + get_db, +) +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + +admin_router = APIRouter(dependencies=[Depends(auth.require(Caller.ADMIN))]) + +@admin_router.post("/admin/update/names") +def update_all_names_to_most_recently_available(db=Depends(get_db)): + logger.info("[FastAPI] ADMIN: Got request to update all existing names") + try: + with db.cursor() as cursor: + cursor.execute( + "SELECT id, anilist_id from seadex_state" + ) + rows = cursor.fetchall() + for id, anilist_id in rows: + anime_name = get_anime_name(anilist_id) + cursor.execute( + "UPDATE seadex_state SET anime_name = %s " + "WHERE id = %s", + (anime_name,id), + ) + db.commit() + logger.info( + "[FastAPI] ADMIN: Updated id=%s, anilist_id=%s with name=%s", + id, anilist_id, anime_name, + ) + except Exception as e: + db.rollback() + logger.exception("[FastAPI] ADMIN: Failed to update all existing names") + raise HTTPException(status_code=500) from e + + logger.info("[FastAPI] ADMIN: Updated %s entries", len(rows)) + return {"status": "ok", "detail": f"Updated {len(rows)} entries"} \ No newline at end of file diff --git a/src/routers/scheduler.py b/src/routers/scheduler.py new file mode 100644 index 0000000..9fb779f --- /dev/null +++ b/src/routers/scheduler.py @@ -0,0 +1,108 @@ + +from fastapi import APIRouter, Depends, HTTPException, Request +from seadex import SeaDexEntry + +from auth_config import Caller, auth +from db_module import ( + get_db, +) +from seadex_client import check_for_update +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") +def check_seadex(request:Request, db=Depends(get_db)): + results = {"checked": 0, "new": 0, "errors": 0} + + with db.cursor() as cursor: + cursor.execute( + "SELECT id, anilist_id, anime_name, last_updated " + "FROM seadex_state WHERE active = 1" + ) + anime = cursor.fetchall() + + logger.debug("[FastAPI] Found %s active anime to check.", len(anime)) + if not anime: + logger.info("[FastAPI] No active Seadex entries") + return results + + with SeaDexEntry() as seadex: + for seadex_state_id, anilist_id, anime_name, last_updated in anime: + results["checked"] += 1 + try: + current = check_for_update(seadex, anilist_id) + if current is None: + logger.debug("[FastAPI] No entry yet for Anime: %s on SeaDex", anime_name) + continue + if current == last_updated: + continue + + logger.info( + "[FastAPI] Anime: %s updated, Anilist ID: %s", + anime_name,anilist_id, + ) + with db.cursor() as cursor: + cursor.execute( + "SELECT user_id, custom_name " + "FROM seadex_state_subscriptions " + "WHERE seadex_state_id = %s AND active = 1", + (seadex_state_id,), + ) + subscribers = cursor.fetchall() + + sent = 0 + failed = 0 + for user_id, custom_name in subscribers: + try: + request.app.state.notifier.send( + title=custom_name or anime_name, + body="SeaDex entry updated", + link=f"https://seadex.moe/{anilist_id}/", + user_id=user_id, + ) + sent += 1 + except NotificationUnavailableError: + failed = len(subscribers) - sent + logger.error("[FastAPI] Notifications backend unavailable, skipping " + "%s remaining user(s) for Anime: %s", + failed, anime_name, + ) + break + except NotificationError as e: + failed += 1 + logger.error( + "[FastAPI] Failed to send notification to user_id=%s for Anime: %s, status=%s", + user_id, anime_name, getattr(e, 'status_code',None), + ) + if subscribers and sent == 0: + results["errors"] += 1 + logger.error("[FastAPI] All notifications failed for Anime: %s, will retry later", anime_name) + continue + + with db.cursor() as cursor: + cursor.execute( + "UPDATE seadex_state SET last_updated = %s WHERE id = %s", + (current, seadex_state_id), + ) + db.commit() + logger.debug("[FastAPI] Notification sent and database updated for Anime: %s", anime_name) + results["new"] += 1 + + except Exception: + db.rollback() + results["errors"] += 1 + logger.exception( + "[FastAPI] Failed for id: %s Anilist: %s", + seadex_state_id, anilist_id, + ) + + if results["errors"]: + logger.warning("[FastAPI] SeaDex check finished with errors: %s", results) + raise HTTPException(status_code=500, detail=results) + + logger.info("[FastAPI] SeaDex check complete: %s", 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..485190a --- /dev/null +++ b/src/routers/user.py @@ -0,0 +1,165 @@ + +from fastapi import APIRouter, Depends, HTTPException +from seadex import SeaDexEntry +from service_auth import UserId + +from anilist_query import get_anime_name +from auth_config import Caller, auth +from db_module import ( + get_db, +) +from models import AnimeCreate, AnimeEntry, AnimeList, AnimeRemove +from seadex_client import check_for_update +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + +user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) + +@user_router.post("/anime/add") +def add_anime( + anime: AnimeCreate, + user_id: UserId, + db=Depends(get_db), + ): + logger.info("[FastAPI] Received request to track anilist_id=%s, user_id=%s", anime.anilist_id, user_id) + + with db.cursor() as cursor: + cursor.execute( + "SELECT id, active FROM seadex_state WHERE anilist_id = %s", + (anime.anilist_id,), + ) + known = cursor.fetchone() + + last_updated = None + with SeaDexEntry() as seadex: + if known is None: + anime_name = get_anime_name(anime.anilist_id) + last_updated = check_for_update(seadex, anime.anilist_id) + else: + anime_name, active = known + if not active: + last_updated = check_for_update(seadex, anime.anilist_id) + + with db.cursor() as cursor: + try: + cursor.execute( + "INSERT INTO seadex_state (anilist_id, anime_name, last_updated) " + "VALUES (%s, %s, %s) " + "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, " + "last_updated = COALESCE(VALUES(last_updated), last_updated)", + (anime.anilist_id, anime_name, last_updated), + ) + seadex_state_id = cursor.lastrowid + + cursor.execute( + "INSERT INTO seadex_state_subscriptions (user_id, seadex_state_id, custom_name) " + "VALUES (%s, %s, %s) " + "ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)", + (user_id, seadex_state_id, anime.custom_name or None), + ) + db.commit() + except Exception as e: + db.rollback() + logger.exception( + "[FastAPI] Failed for anilist_id=%s, name=%s, user_id=%s", + anime.anilist_id, anime_name, user_id, + ) + raise HTTPException(status_code=500) from e + + logger.info( + "[FastAPI] Tracking anilist_id=%s name=%r for user_id=%s (state_id=%s)", + anime.anilist_id, anime_name, user_id, seadex_state_id, + ) + return {"status": "ok", "detail": f"Tracking name={anime.custom_name or anime_name}, Anilist ID={anime.anilist_id}"} + +@user_router.post("/anime/delete") +def remove_anime( + anime: AnimeRemove, + user_id: UserId, + db=Depends(get_db) +): + logger.info( + "[FastAPI] Received request to deactivate anilist_id=%s for user_id=%s", + anime.anilist_id, user_id, + ) + with db.cursor() as cursor: + try: + cursor.execute( + "SELECT id FROM seadex_state " + "WHERE anilist_id = %s FOR UPDATE", + (anime.anilist_id,), + ) + known = cursor.fetchone() + + if known is None: + db.rollback() + logger.info("[FastAPI] Anime anilist_id=%s is not tracked, nothing to do", anime.anilist_id) + return {"status": "ok", "detail": f"Stopped tracking Anilist ID={anime.anilist_id}"} + + seadex_state_id = known[0] + cursor.execute( + "UPDATE seadex_state_subscriptions SET active = 0 " + "WHERE user_id = %s AND seadex_state_id = %s", + (user_id, seadex_state_id), + ) + + cursor.execute( + "SELECT 1 from seadex_state_subscriptions " + "WHERE seadex_state_id = %s AND active = 1 LIMIT 1", + (seadex_state_id,), + ) + + if cursor.fetchone() is None: + cursor.execute( + "UPDATE seadex_state SET active = 0 WHERE id = %s", + (seadex_state_id,), + ) + logger.info("[FastAPI] No active subscribers left, deactivated state_id=%s", seadex_state_id) + + db.commit() + except Exception as e: + db.rollback() + logger.exception( + "[FastAPI] Failed to unsubscribe anime anilist_id=%s for user_id=%s", + anime.anilist_id, user_id, + ) + raise HTTPException(status_code=500) from e + + logger.info( + "[FastAPI] Successfully deactivated tracking for anilist_id=%s for user_id=%s", + anime.anilist_id, user_id, + ) + return {"status": "ok", "detail": f"Stopped tracking Anilist ID={anime.anilist_id}"} + +@user_router.get("/anime/list", response_model=AnimeList) +def list_active_and_inactive_anime( + user_id: UserId, + db=Depends(get_db), + ): + logger.info("[FastAPI] Received request to list anime for User ID = %s", user_id) + + try: + with db.cursor() as cursor: + cursor.execute( + "SELECT s.custom_name AS custom_name, st.anime_name AS anime_name, " + "st.anilist_id, s.active, st.last_updated from seadex_state_subscriptions AS s " + "JOIN seadex_state AS st on s.seadex_state_id = st.id " + "WHERE s.user_id = %s ORDER BY st.anilist_id", + (user_id,), + ) + rows = cursor.fetchall() + + anime_list=AnimeList() + for custom_name, anime_name, anilist_id, is_active, last_updated in rows: + entry = AnimeEntry( + name=custom_name or anime_name, + anilist_id=anilist_id, + last_updated=None if last_updated is None else last_updated, + ) + (anime_list.active if is_active else anime_list.inactive).append(entry) + except Exception as e: + logger.exception("[FastAPI] Failed to list anime") + raise HTTPException(status_code=500) from e + + return anime_list \ No newline at end of file diff --git a/src/seadex_client.py b/src/seadex_client.py new file mode 100644 index 0000000..ffa1c6a --- /dev/null +++ b/src/seadex_client.py @@ -0,0 +1,14 @@ +from datetime import UTC, datetime + +from seadex import EntryNotFoundError, SeaDexEntry + + +def to_iso(dt: datetime) -> str: + return dt.astimezone(UTC).isoformat(timespec="microseconds") + + +def check_for_update(seadex: SeaDexEntry, anilist_id: int) -> str | None: + try: + return to_iso(seadex.from_id(anilist_id).updated_at) + except EntryNotFoundError: + return None \ No newline at end of file