Moved routes to their own modules, added path to list subscribed repositories

src/main.py

@@ -3,29 +3,16 @@ import os
 from contextlib import asynccontextmanager
 
 import uvicorn
-from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
+from fastapi import Depends, FastAPI
-from service_auth import UserId
 
 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 handlers import register_handlers
 from metrics_server import metrics_api
-from models import RepositoryType, RepositoryUpdate
+from routers.scheduler import scheduler_router
-from repositories import (
+from routers.user import user_router
-	fetch_latest,
-	reset_missing_checks,
-	unsubscribe_every_subscriber_from_repository,
-	validate_repository_combination_and_return_latest_tag,
-)
-from repository_errors import (
-	ApiAuthenticationFailure,
-	RateLimitedError,
-	RepositoryApiError,
-	RepositoryNotFoundError,
-	RepositoryTagNotFoundError,
-)
 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__)
@@ -65,218 +52,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/{repository_type}")
-def update_watched_repositories(
-	request: Request,
-	repository_type: RepositoryType,
-	db=Depends(get_db),
-	):
-	results = {"checked": 0, "new": 0, "errors": 0}
-	with db.cursor() as cursor:
-		cursor.execute(
-			"SELECT id, org, repo, tag, latest_entry, missing_checks "
-			"FROM watched_repositories WHERE type = %s AND active = 1",
-			(repository_type.value,),
-		)
-		repositories = cursor.fetchall()
-	logger.debug(f"[FastAPI] Found {len(repositories)} active {repository_type.value} repositories to check.")
-
-	for repository_id, org, repo, tag, latest_entry, missing_checks in repositories:
-		results["checked"] += 1
-		try:
-			online_version = fetch_latest(repository_type, org, repo, tag)
-		except RateLimitedError as e:
-			logger.warning(f"[FastAPI] Rate limited, stopping the {repository_type.value} check: {e}")
-			results["errors"] += 1
-			break
-		except ApiAuthenticationFailure as e:
-			logger.error(f"[FastAPI] {e}")
-			results["errors"] += 1
-			break
-		except (RepositoryNotFoundError, RepositoryTagNotFoundError) as e:
-			logger.warning(f"[FastAPI] {e}, deactivating repository ID {repository_id}")
-			unsubscribe_every_subscriber_from_repository(db, request.app.state.notifier, repository_id, org, repo, tag)
-			continue
-		except RepositoryApiError as e:
-			logger.warning(f"[FastAPI] Lookup failed for ID {repository_id}: {e}")
-			results["errors"] += 1
-			continue
-		except Exception:
-			logger.exception(f"[FastAPI] Unexpected error for ID {repository_id} ({org}/{repo}:{tag})")
-			results["errors"] += 1
-			continue
-
-		if missing_checks:
-			reset_missing_checks(db, repository_id)
-		if latest_entry == online_version:
-			continue
-
-		try:
-
-			logger.info(f"[FastAPI] New version detected for org={org}, repo={repo}, tag={tag}")
-
-			with db.cursor() as cursor:
-				cursor.execute(
-					"SELECT user_id, custom_name "
-					"FROM watched_repositories_subscriptions "
-					"WHERE repository_id = %s AND active = 1",
-					(repository_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 f"{org}/{repo}",
-						body=f"New version {online_version} for tag {tag}",
-						category="utility",
-						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 repo={org}/{repo}"
-					)
-					break
-				except NotificationError as e:
-					failed += 1
-					logger.error(
-						f"[FastAPI] Failed to send notification to user_id={user_id} for repo={org}/{repo} "
-						f"status={getattr(e, 'status_code', None)}"
-					)
-
-			if subscribers and sent == 0:
-				results["errors"] += 1
-				logger.error(f"[FastAPI] All notifications failed for ID:{repository_id}, will retry later")
-				continue
-
-			with db.cursor() as cursor:
-				cursor.execute(
-					"UPDATE watched_repositories SET latest_entry = %s WHERE id = %s",
-					(online_version, repository_id),
-				)
-			db.commit()
-			logger.debug(f"[FastAPI] Notification sent and database updated for repository ID {repository_id}")
-			results["new"] += 1
-
-		except Exception as e:
-			db.rollback()
-			results["errors"] += 1
-			logger.error(
-				f"[FastAPI] Failed for ID {repository_id} (org={org}, repo={repo}, tag={tag}): {e}",
-				exc_info=True,
-			)
-
-	if results["errors"]:
-		logger.warning(f"[FastAPI] {repository_type.value} check finished with errors: {results}")
-		raise HTTPException(status_code=500, detail=results)
-	logger.info(f"[FastAPI] {repository_type.value} check complete: {results}")
-	return results
-
-@user_router.post("/add/repository")
-def add_repository_subscription(
-	user_id: UserId,
-	repository: RepositoryUpdate,
-	db=Depends(get_db),
-):
-	name = f"{repository.org}/{repository.repo}:{repository.tag}"
-	logger.info(f"[FastAPI] Subscribe request: {repository.type.value} {name}, user_id={user_id}")
-
-	with db.cursor() as cursor:
-		cursor.execute(
-			"SELECT id, active FROM watched_repositories "
-			"WHERE type = %s AND org = %s AND repo = %s AND tag = %s",
-			(repository.type.value, repository.org, repository.repo, repository.tag),
-		)
-		known = cursor.fetchone()
-
-	latest_entry = None
-	if known is None or not known[1]:
-		latest_entry = validate_repository_combination_and_return_latest_tag(repository)
-
-	try:
-		with db.cursor() as cursor:
-			cursor.execute(
-				"INSERT INTO watched_repositories (type, org, repo, tag, latest_entry) "
-				"VALUES (%s, %s, %s, %s, %s) "
-				"ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, missing_checks = 0, "
-				"latest_entry = COALESCE(VALUES(latest_entry), latest_entry)",
-				(repository.type.value, repository.org, repository.repo, repository.tag, latest_entry),
-			)
-			repository_id = cursor.lastrowid
-
-			cursor.execute(
-				"INSERT INTO watched_repositories_subscriptions (user_id, repository_id, custom_name) "
-				"VALUES (%s, %s, %s) "
-				"ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)",
-				(user_id, repository_id, repository.custom_name),
-			)
-		db.commit()
-	except Exception as e:
-		db.rollback()
-		logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {name}")
-		raise HTTPException(status_code=500) from e
-
-	logger.info(f"[FastAPI] user_id={user_id} subscribed to repository ID {repository_id} ({name})")
-	return {"status": "ok", "detail": "Subscribed to repository", "name": name}
-
-@user_router.post("/delete/repository")
-def delete_repository_subscription(
-	user_id: UserId,
-	repository: RepositoryUpdate,
-	db=Depends(get_db),
-):
-	name = f"{repository.org}/{repository.repo}:{repository.tag}"
-	logger.info(f"[FastAPI] Unsubscribe request: {repository.type.value} {name}, user_id={user_id}")
-
-	try:
-		with db.cursor() as cursor:
-			cursor.execute(
-				"SELECT id FROM watched_repositories "
-				"WHERE type = %s AND org = %s AND repo = %s AND tag = %s FOR UPDATE",
-				(repository.type.value, repository.org, repository.repo, repository.tag),
-			)
-			known = cursor.fetchone()
-
-			if known is None:
-				db.rollback()
-				logger.info(f"[FastAPI] Repository {repository.type.value} {name} is not tracked, nothing to do")
-				return {"status":"ok","detail":"Unsubscribed from repository","name":name}
-
-			repository_id = known[0]
-			cursor.execute(
-				"UPDATE watched_repositories_subscriptions SET active = 0 "
-				"WHERE user_id = %s AND repository_id = %s",
-				(user_id, repository_id),
-			)
-
-			cursor.execute(
-				"SELECT 1 from watched_repositories_subscriptions "
-				"WHERE repository_id = %s AND active = 1 LIMIT 1 FOR SHARE",
-				(repository_id,),
-			)
-			if cursor.fetchone() is None:
-				cursor.execute(
-					"UPDATE watched_repositories SET active = 0 WHERE id = %s",
-					(repository_id,),
-				)
-			db.commit()
-	except Exception as e:
-		db.rollback()
-		logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {name}")
-		raise HTTPException(status_code=500) from e
-
-	logger.info(f"[FastAPI] Unsubscribe request successful: {repository.type.value} {name}, user_id={user_id}")
-	return {"status":"ok","detail":"Unsubscribed from repository","name":name}
-
 api.include_router(user_router)
 api.include_router(scheduler_router)
 

src/models.py

@@ -1,7 +1,7 @@
 import string
 from enum import StrEnum
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, computed_field, field_validator
 
 from repository_errors import InvalidOrganisationError, InvalidRepositoryError, InvalidTagError
 
@@ -10,6 +10,25 @@ _MAX_ORG = 39
 _MAX_REPO = 255
 _MAX_TAG = 128
 
+class SubscribedRepository(BaseModel):
+	type: str
+	repository_name: str
+	tag: str
+	latest_entry: str
+
+class Counts(BaseModel):
+	active: int
+	inactive: int
+
+class SubscribedRepositoryList(BaseModel):
+	active: list[SubscribedRepository] = Field(default_factory=list)
+	inactive: list[SubscribedRepository] = Field(default_factory=list)
+
+	@computed_field
+	@property
+	def counts(self) -> Counts:
+		return Counts(active=len(self.active),inactive=len(self.inactive))
+
 class RepositoryType(StrEnum):
 	GITHUB = "github"
 	DOCKERHUB = "dockerhub"

src/routers/scheduler.py

@@ -0,0 +1,137 @@
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+
+from auth_config import Caller, auth
+from db_module import get_db
+from models import RepositoryType
+from repositories import (
+	fetch_latest,
+	reset_missing_checks,
+	unsubscribe_every_subscriber_from_repository,
+)
+from repository_errors import (
+	ApiAuthenticationFailure,
+	RateLimitedError,
+	RepositoryApiError,
+	RepositoryNotFoundError,
+	RepositoryTagNotFoundError,
+)
+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/{repository_type}")
+def update_watched_repositories(
+	request: Request,
+	repository_type: RepositoryType,
+	db=Depends(get_db),
+	):
+	results = {"checked": 0, "new": 0, "errors": 0}
+	with db.cursor() as cursor:
+		cursor.execute(
+			"SELECT id, org, repo, tag, latest_entry, missing_checks "
+			"FROM watched_repositories WHERE type = %s AND active = 1",
+			(repository_type.value,),
+		)
+		repositories = cursor.fetchall()
+	logger.debug(f"[FastAPI] Found {len(repositories)} active {repository_type.value} repositories to check.")
+
+	for repository_id, org, repo, tag, latest_entry, missing_checks in repositories:
+		results["checked"] += 1
+		try:
+			online_version = fetch_latest(repository_type, org, repo, tag)
+		except RateLimitedError as e:
+			logger.warning(f"[FastAPI] Rate limited, stopping the {repository_type.value} check: {e}")
+			results["errors"] += 1
+			break
+		except ApiAuthenticationFailure as e:
+			logger.error(f"[FastAPI] {e}")
+			results["errors"] += 1
+			break
+		except (RepositoryNotFoundError, RepositoryTagNotFoundError) as e:
+			logger.warning(f"[FastAPI] {e}, deactivating repository ID {repository_id}")
+			unsubscribe_every_subscriber_from_repository(db, request.app.state.notifier, repository_id, org, repo, tag)
+			continue
+		except RepositoryApiError as e:
+			logger.warning(f"[FastAPI] Lookup failed for ID {repository_id}: {e}")
+			results["errors"] += 1
+			continue
+		except Exception:
+			logger.exception(f"[FastAPI] Unexpected error for ID {repository_id} ({org}/{repo}:{tag})")
+			results["errors"] += 1
+			continue
+
+		if missing_checks:
+			reset_missing_checks(db, repository_id)
+		if latest_entry == online_version:
+			continue
+
+		try:
+
+			logger.info(f"[FastAPI] New version detected for org={org}, repo={repo}, tag={tag}")
+
+			with db.cursor() as cursor:
+				cursor.execute(
+					"SELECT user_id, custom_name "
+					"FROM watched_repositories_subscriptions "
+					"WHERE repository_id = %s AND active = 1",
+					(repository_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 f"{org}/{repo}",
+						body=f"New version {online_version} for tag {tag}",
+						category="utility",
+						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 repo={org}/{repo}"
+					)
+					break
+				except NotificationError as e:
+					failed += 1
+					logger.error(
+						f"[FastAPI] Failed to send notification to user_id={user_id} for repo={org}/{repo} "
+						f"status={getattr(e, 'status_code', None)}"
+					)
+
+			if subscribers and sent == 0:
+				results["errors"] += 1
+				logger.error(f"[FastAPI] All notifications failed for ID:{repository_id}, will retry later")
+				continue
+
+			with db.cursor() as cursor:
+				cursor.execute(
+					"UPDATE watched_repositories SET latest_entry = %s WHERE id = %s",
+					(online_version, repository_id),
+				)
+			db.commit()
+			logger.debug(f"[FastAPI] Notification sent and database updated for repository ID {repository_id}")
+			results["new"] += 1
+
+		except Exception as e:
+			db.rollback()
+			results["errors"] += 1
+			logger.error(
+				f"[FastAPI] Failed for ID {repository_id} (org={org}, repo={repo}, tag={tag}): {e}",
+				exc_info=True,
+			)
+
+	if results["errors"]:
+		logger.warning(f"[FastAPI] {repository_type.value} check finished with errors: {results}")
+		raise HTTPException(status_code=500, detail=results)
+	logger.info(f"[FastAPI] {repository_type.value} check complete: {results}")
+	return results

src/routers/user.py

@@ -0,0 +1,147 @@
+
+from fastapi import APIRouter, Depends, HTTPException
+from service_auth import UserId
+
+from auth_config import Caller, auth
+from db_module import get_db
+from models import RepositoryUpdate, SubscribedRepository, SubscribedRepositoryList
+from repositories import (
+	validate_repository_combination_and_return_latest_tag,
+)
+from simple_logger_handler import setup_logger
+
+logger = setup_logger(__name__)
+
+user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
+
+
+@user_router.post("/repository/add")
+def add_repository_subscription(
+	user_id: UserId,
+	repository: RepositoryUpdate,
+	db=Depends(get_db),
+):
+	name = f"{repository.org}/{repository.repo}:{repository.tag}"
+	logger.info(f"[FastAPI] Subscribe request: {repository.type.value} {name}, user_id={user_id}")
+
+	with db.cursor() as cursor:
+		cursor.execute(
+			"SELECT id, active FROM watched_repositories "
+			"WHERE type = %s AND org = %s AND repo = %s AND tag = %s",
+			(repository.type.value, repository.org, repository.repo, repository.tag),
+		)
+		known = cursor.fetchone()
+
+	latest_entry = None
+	if known is None or not known[1]:
+		latest_entry = validate_repository_combination_and_return_latest_tag(repository)
+
+	try:
+		with db.cursor() as cursor:
+			cursor.execute(
+				"INSERT INTO watched_repositories (type, org, repo, tag, latest_entry) "
+				"VALUES (%s, %s, %s, %s, %s) "
+				"ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, missing_checks = 0, "
+				"latest_entry = COALESCE(VALUES(latest_entry), latest_entry)",
+				(repository.type.value, repository.org, repository.repo, repository.tag, latest_entry),
+			)
+			repository_id = cursor.lastrowid
+
+			cursor.execute(
+				"INSERT INTO watched_repositories_subscriptions (user_id, repository_id, custom_name) "
+				"VALUES (%s, %s, %s) "
+				"ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)",
+				(user_id, repository_id, repository.custom_name),
+			)
+		db.commit()
+	except Exception as e:
+		db.rollback()
+		logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {name}")
+		raise HTTPException(status_code=500) from e
+
+	logger.info(f"[FastAPI] user_id={user_id} subscribed to repository ID {repository_id} ({name})")
+	return {"status": "ok", "detail": "Subscribed to repository", "name": name}
+
+@user_router.post("/repository/delete")
+def delete_repository_subscription(
+	user_id: UserId,
+	repository: RepositoryUpdate,
+	db=Depends(get_db),
+):
+	name = f"{repository.org}/{repository.repo}:{repository.tag}"
+	logger.info(f"[FastAPI] Unsubscribe request: {repository.type.value} {name}, user_id={user_id}")
+
+	try:
+		with db.cursor() as cursor:
+			cursor.execute(
+				"SELECT id FROM watched_repositories "
+				"WHERE type = %s AND org = %s AND repo = %s AND tag = %s FOR UPDATE",
+				(repository.type.value, repository.org, repository.repo, repository.tag),
+			)
+			known = cursor.fetchone()
+
+			if known is None:
+				db.rollback()
+				logger.info(f"[FastAPI] Repository {repository.type.value} {name} is not tracked, nothing to do")
+				return {"status":"ok","detail":"Unsubscribed from repository","name":name}
+
+			repository_id = known[0]
+			cursor.execute(
+				"UPDATE watched_repositories_subscriptions SET active = 0 "
+				"WHERE user_id = %s AND repository_id = %s",
+				(user_id, repository_id),
+			)
+
+			cursor.execute(
+				"SELECT 1 from watched_repositories_subscriptions "
+				"WHERE repository_id = %s AND active = 1 LIMIT 1 FOR SHARE",
+				(repository_id,),
+			)
+			if cursor.fetchone() is None:
+				cursor.execute(
+					"UPDATE watched_repositories SET active = 0 WHERE id = %s",
+					(repository_id,),
+				)
+			db.commit()
+	except Exception as e:
+		db.rollback()
+		logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {name}")
+		raise HTTPException(status_code=500) from e
+
+	logger.info(f"[FastAPI] Unsubscribe request successful: {repository.type.value} {name}, user_id={user_id}")
+	return {"status":"ok","detail":"Unsubscribed from repository","name":name}
+
+@user_router.post("/repository/list")
+def list_subscribed_repositories(
+	user_id: UserId,
+	db = Depends(get_db),
+):
+	logger.info(f"[FastAPI] Received request to list repositories for user_id={user_id}")
+	try:
+		with db.cursor() as cursor:
+			cursor.execute(
+				"select w.type, w.org, w.repo, w.tag, w.latest_entry, s.custom_name, s.active from watched_repositories "
+				"AS w JOIN watched_repositories_subscriptions AS s ON w.id = s.repository_id "
+				"WHERE s.user_id = %s ORDER BY w.type",
+				(user_id,),
+			)
+			rows = cursor.fetchall()
+
+		repository_list = SubscribedRepositoryList()
+		for type, org, repo, tag, latest_entry, custom_name, is_active in rows:
+			entry = SubscribedRepository(
+				type = type,
+				repository_name = custom_name or f"{org}/{repo}",
+				tag = tag,
+				latest_entry = latest_entry,
+			)
+			(repository_list.active if is_active else repository_list.inactive).append(entry)
+
+	except Exception as e:
+		logger.exception("[FastAPI] Failed to list repository")
+		raise HTTPException(status_code=500) from e
+
+	logger.info(f"[FastAPI] Listed all repositories for user_id={user_id}")
+	return repository_list
+
+