notifier/service-seadex
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added API key requirement to main paths and support for different users #1

Merged
florian merged 1 commits from feature/user-id-support into main 2026-10-29 17:44:49 +00:00
Conversation 0 Commits 1 Files Changed 5
+91 -44
5 changed files with 91 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

database.sql
+12
View File
@@ -0,0 +1,12 @@
CREATE TABLE `seadex_state` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`user_id` bigint(20) NOT NULL,
`anilist_id` int(10) unsigned NOT NULL,
`anime_name` varchar(255) NOT NULL,
`last_updated` varchar(40) DEFAULT NULL,
`active` tinyint(1) NOT NULL DEFAULT '1',
`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uq_user_anilist` (`user_id`,`anilist_id`),
CONSTRAINT `fk_seadex_state_user` FOREIGN KEY (`user_id`) REFERENCES `users` (`user_id`) ON DELETE CASCADE
)
requirements.txt
+2 -1
View File
@@ -12,8 +12,9 @@ httpcore==1.0.9
httpx==0.28.1
idna==3.20
lib-db-module==0.3.0
lib-secret-manager==0.3.1
lib-secret-manager==0.3.6
lib-send-notifications==0.1.0
lib-service-auth==0.1.0
msgspec==0.22.0
mysql-connector-python==9.4.0
opentelemetry-api==1.45.1
src/auth_config.py
+13
View File
@@ -0,0 +1,13 @@
from enum import StrEnum
from service_auth import KeySource, ServiceAuth
class Caller(StrEnum):
GATEWAY = "gateway"
SCHEDULER = "scheduler"
auth = ServiceAuth({
Caller.GATEWAY: KeySource("GATEWAY_KEY", "/etc/secrets/gateway_key"),
Caller.SCHEDULER: KeySource("SCHEDULER_KEY", "/etc/secrets/scheduler_key"),
})
src/main.py
+58 -41
View File
@@ -4,11 +4,13 @@ from contextlib import asynccontextmanager
from datetime import UTC, datetime
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
from pydantic import BaseModel, Field, computed_field
from seadex import EntryNotFoundError, SeaDexEntry
from service_auth import UserId
from anilist_query import AniListError, AnimeNotFoundError, MissingAnimeName, get_anime_name
from auth_config import Caller, auth
from db_module import (
close_connection_pool,
create_connection_pool,
@@ -45,7 +47,7 @@ class AnimeList(BaseModel):
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("[App] Starting application...")
app.state.notifier = NotificationClient()
app.state.notifier = NotificationClient()
logger.info("[DB] Creating MySQL connection pool...")
create_connection_pool("seadexPool")
@@ -70,9 +72,13 @@ api = FastAPI(
title="Service SeaDex",
description="Queries SeaDex for new torrent information",
version="1.0.0",
lifespan=lifespan
lifespan=lifespan,
dependencies=[Depends(auth.authenticate)],
)
user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
@api.middleware("http")
async def prometheus_middleware(request: Request, call_next):
logger.debug(f"[Metrics] Incoming request: {request.method} {request.url.path}")
@@ -101,22 +107,25 @@ def check_for_update(seadex: SeaDexEntry, anilist_id: int) -> str | None:
return None
@api.post("/update")
@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 "
"SELECT id, anilist_id, anime_name, last_updated, user_id "
"FROM seadex_state WHERE active = 1"
)
anime = cursor.fetchall()
cache: dict[int, str | None] = {}
with SeaDexEntry() as seadex:
for row_id, anilist_id, anime_name, last_updated in anime:
for row_id, anilist_id, anime_name, last_updated, user_id in anime:
results["checked"] += 1
try:
current = check_for_update(seadex, anilist_id)
if anilist_id not in cache:
cache[anilist_id] = check_for_update(seadex, anilist_id)
current = cache[anilist_id]
if current is None:
logger.debug(f"[FastAPI] No entry yet for {anime_name} on SeaDex")
@@ -124,31 +133,32 @@ def check_seadex(request:Request, db=Depends(get_db)):
if current != last_updated:
logger.info(
f"[FastAPI] {anime_name} updated, Anilist ID: {anilist_id}"
f"[FastAPI] {anime_name} updated, Anilist ID: {anilist_id}, User ID: {user_id}"
)
request.app.state.notifier.send(
title=anime_name,
body="SeaDex entry updated",
link=f"https://seadex.moe/{anilist_id}/",
user_id=user_id,
)
with db.cursor() as cursor:
cursor.execute(
"UPDATE seadex_state SET last_updated = %s WHERE id = %s",
(current, row_id),
"UPDATE seadex_state SET last_updated = %s WHERE id = %s AND user_id = %s",
(current, row_id, user_id),
)
db.commit()
logger.debug(
f"[FastAPI] Notified and updated db for id={row_id}, "
f"Anilist ID={anilist_id}"
f"Anilist ID={anilist_id}, User ID: {user_id}"
)
results["new"] += 1
except Exception:
db.rollback()
results["errors"] += 1
logger.exception(f"[FastAPI] Failed for id={row_id} Anilist={anilist_id}",)
logger.exception(f"[FastAPI] Failed for id={row_id} Anilist={anilist_id}, User ID: {user_id}",)
if results["errors"]:
logger.warning(f"[FastAPI] SeaDex check finished with errors: {results}")
@@ -157,9 +167,9 @@ def check_seadex(request:Request, db=Depends(get_db)):
logger.info(f"[FastAPI] SeaDex check complete: {results}")
return results
@api.post("/add/anime")
def add_anime(anime: AnimeCreate, db=Depends(get_db)):
logger.info(f"[FastAPI] Received request to track anilist_id={anime.anilist_id}")
@user_router.post("/add/anime")
def add_anime(anime: AnimeCreate, user_id: UserId, db=Depends(get_db)):
logger.info(f"[FastAPI] Received request to track anilist_id={anime.anilist_id}, user_id={user_id}")
try:
anime_name = get_anime_name(anime.anilist_id)
@@ -175,16 +185,16 @@ def add_anime(anime: AnimeCreate, db=Depends(get_db)):
with db.cursor() as cursor:
try:
cursor.execute(
"SELECT id FROM seadex_state WHERE anilist_id = %s",
(anime.anilist_id,),
"SELECT id FROM seadex_state WHERE anilist_id = %s AND user_id = %s",
(anime.anilist_id, user_id),
)
if cursor.fetchone():
logger.debug("[FastAPI] Found existing entry, updating name and setting to active")
cursor.execute(
"UPDATE seadex_state SET anime_name = %s, active = 1 "
"WHERE anilist_id = %s",
(anime_name, anime.anilist_id),
"WHERE anilist_id = %s AND user_id=%s",
(anime_name, anime.anilist_id, user_id),
)
db.commit()
return {
@@ -195,12 +205,12 @@ def add_anime(anime: AnimeCreate, db=Depends(get_db)):
with SeaDexEntry() as seadex:
current = check_for_update(seadex, anime.anilist_id)
logger.debug("[FastAPI] Inserting new anime with last_updated={current}")
logger.debug(f"[FastAPI] Inserting new anime with last_updated={current}")
cursor.execute(
"INSERT INTO seadex_state (anilist_id, anime_name, last_updated) "
"VALUES (%s, %s, %s)",
(anime.anilist_id, anime_name, current),
"INSERT INTO seadex_state (anilist_id, anime_name, last_updated, user_id) "
"VALUES (%s, %s, %s, %s)",
(anime.anilist_id, anime_name, current, user_id),
)
db.commit()
return {
@@ -211,27 +221,27 @@ def add_anime(anime: AnimeCreate, db=Depends(get_db)):
except Exception as e:
db.rollback()
logger.exception(f"[FastAPI] Failed for anilist_id={anime.anilist_id} name={anime_name}")
logger.exception(f"[FastAPI] Failed for anilist_id={anime.anilist_id}, name={anime_name}, user_id={user_id}")
raise HTTPException(status_code=500) from e
@api.post("/delete/anime")
def remove_anime(anime: AnimeCreate, db=Depends(get_db)):
logger.info(f"[FastAPI] Received request to deactivate tracking for anilist_id={anime.anilist_id}")
@user_router.post("/delete/anime")
def remove_anime(anime: AnimeCreate, user_id: UserId, db=Depends(get_db)):
logger.info(f"[FastAPI] Received request to deactivate tracking for anilist_id={anime.anilist_id}, user_id={user_id}")
with db.cursor() as cursor:
try:
cursor.execute(
"SELECT active FROM seadex_state WHERE anilist_id = %s",
(anime.anilist_id,),
"SELECT active FROM seadex_state WHERE anilist_id = %s AND user_id = %s",
(anime.anilist_id, user_id),
)
row = cursor.fetchone()
if row and row[0]:
cursor.execute(
"UPDATE seadex_state SET active = 0 "
"WHERE anilist_id = %s",
(anime.anilist_id,),
"WHERE anilist_id = %s AND user_id = %s",
(anime.anilist_id, user_id),
)
db.commit()
logger.debug(f"[FastAPI] Successfully deactivated tracking for anilist_id={anime.anilist_id}")
logger.debug(f"[FastAPI] Successfully deactivated tracking for anilist_id={anime.anilist_id}, user_id={user_id}")
return {
"status": "ok",
"detail": f"Deactivated entry {anime.anilist_id}",
@@ -239,28 +249,30 @@ def remove_anime(anime: AnimeCreate, db=Depends(get_db)):
except Exception as e:
db.rollback()
logger.exception(f"[FastAPI] Failed removeing tracking for id={anime.anilist_id}")
logger.exception(f"[FastAPI] Failed removeing tracking for id={anime.anilist_id}, user_id={user_id}")
raise HTTPException(status_code=500) from e
if row is None:
logger.debug(f"[FastAPI] No tracked entry for anilist_id={anime.anilist_id}")
logger.debug(f"[FastAPI] No tracked entry for anilist_id={anime.anilist_id}, user_id={user_id}")
raise HTTPException(status_code=404, detail="No tracked entry for this AniList ID")
if not row[0]:
logger.debug(f"[FastAPI] Tracking was already deactivated for anilist_id={anime.anilist_id}")
logger.debug(f"[FastAPI] Tracking was already deactivated for anilist_id={anime.anilist_id}, user_id={user_id}")
return {"status": "ok", "detail": f"Entry {anime.anilist_id} was already inactive"}
@api.get("/list/anime", response_model=AnimeList)
def list_active_and_inactive_anime(db=Depends(get_db)):
logger.info(f"[FastAPI] Received request to list anime")
@user_router.get("/list/anime", response_model=AnimeList)
def list_active_and_inactive_anime(user_id: UserId, db=Depends(get_db)):
logger.info(f"[FastAPI] Received request to list anime for User ID = {user_id}")
with db.cursor() as cursor:
try:
cursor.execute(
"SELECT anime_name, active, last_updated " \
"FROM seadex_state ORDER BY anime_name"
"FROM seadex_state WHERE user_id = %s "
"ORDER BY anime_name",
(user_id,),
)
rows = cursor.fetchall()
except Exception as e:
logger.exception(f"[FastAPI] Failed to list anime")
logger.exception("[FastAPI] Failed to list anime")
raise HTTPException(status_code=500) from e
anime_list=AnimeList ()
@@ -272,11 +284,16 @@ def list_active_and_inactive_anime(db=Depends(get_db)):
(anime_list.active if is_active else anime_list.inactive).append(entry)
return anime_list
api.include_router(user_router)
api.include_router(scheduler_router)
async def start_servers():
logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...")
logger.info(f"[Server] Build commit id={os.getenv("COMMIT_ID","unknown")}")
auth.load_keys(api, Caller.GATEWAY, Caller.SCHEDULER)
cleanup_secret_files(auth.key_paths(Caller.GATEWAY, Caller.SCHEDULER))
config_main = uvicorn.Config(api, host="0.0.0.0", port=5000, log_level=LOG_LEVEL.lower(),log_config=None)
config_metrics = uvicorn.Config(metrics_api, host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(),log_config=None)
src/metrics_server.py
+6 -2
View File
@@ -1,10 +1,14 @@
from fastapi import FastAPI, Response
from prometheus_client import CONTENT_TYPE_LATEST, Counter, generate_latest
metrics_api = FastAPI(title="Metrics Server", description="Prometheus metrics endpoint")
metrics_api = FastAPI(
title="Metrics Server",
description="Prometheus metrics endpoint",
)
REQUEST_COUNTER = Counter("http_requests_total", "Total HTTP Requests", ["method", "endpoint", "status"])
@metrics_api.get("/metrics")
async def metrics():
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)