Moved routes to their own modules

This commit is contained in:
florian committed 2026-11-05 12:45:14 +01:00
1 parent 3b649e8862
commit cd0278d8a8
6 files changed
+333 -295

No files matched your search

+4 -295
View File
@@ -1,32 +1,26 @@
import asyncio import asyncio
import os import os
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from datetime import UTC, datetime
import uvicorn import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request from fastapi import Depends, FastAPI
from seadex import EntryNotFoundError, SeaDexEntry
from service_auth import UserId
from anilist_query import get_anime_name
from auth_config import Caller, auth from auth_config import Caller, auth
from db_module import ( from db_module import (
close_connection_pool, close_connection_pool,
create_connection_pool, create_connection_pool,
get_db,
start_healthcheck_thread, start_healthcheck_thread,
) )
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 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 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 from simple_logger_handler import LOG_LEVEL, setup_logger
logger = setup_logger(__name__) logger = setup_logger(__name__)
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
logger.info("[App] Starting application...") logger.info("[App] Starting application...")
@@ -60,291 +54,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))])
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(user_router)
api.include_router(scheduler_router) api.include_router(scheduler_router)
api.include_router(admin_router) api.include_router(admin_router)
View File
Whitespace-only changes.
+42
View File
@@ -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"}
+108
View File
@@ -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
+165
View File
@@ -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
+14
View File
@@ -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