- Added general authentication that seperates user calles from a gateway and cron scheduler updates

- Added notifications being seperate by user_id
- Added user paths to subscribe/unsubscribe from a repository
- Cleaned up main.py by moving handlers to its own module
This commit is contained in:
florian committed 2026-11-03 10:41:14 +01:00
1 parent 7181cdbd1b
commit e06a9f5224
5 files changed
+483 -86

No files matched your search

+28
View File
@@ -0,0 +1,28 @@
CREATE TABLE `watched_repositories` (
`id` int unsigned NOT NULL AUTO_INCREMENT,
`type` enum('github','dockerhub') COLLATE utf8mb4_unicode_ci NOT NULL,
`org` varchar(39) COLLATE utf8mb4_unicode_ci NOT NULL,
`repo` varchar(100) COLLATE utf8mb4_unicode_ci NOT NULL,
`tag` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
`latest_entry` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
`active` tinyint(1) NOT NULL DEFAULT '1',
`missing_checks` tinyint unsigned NOT NULL DEFAULT '0',
`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uq_repository` (`type`,`org`,`repo`,`tag`),
KEY `idx_active_type` (`active`,`type`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE `watched_repositories_subscriptions` (
`id` int unsigned NOT NULL AUTO_INCREMENT,
`user_id` bigint NOT NULL,
`repository_id` int unsigned NOT NULL,
`custom_name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
`active` tinyint(1) NOT NULL DEFAULT '1',
`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uq_user_repository` (`user_id`,`repository_id`),
KEY `idx_repository_active` (`repository_id`,`active`),
CONSTRAINT `fk_wrs_repository` FOREIGN KEY (`repository_id`) REFERENCES `watched_repositories` (`id`) ON DELETE CASCADE,
CONSTRAINT `fk_wrs_user` FOREIGN KEY (`user_id`) REFERENCES `users` (`user_id`) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci ;
+66
View File
@@ -0,0 +1,66 @@
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from metrics_server import REQUEST_COUNTER
from repository_errors import RepositoryError
from simple_logger_handler import setup_logger
logger = setup_logger(__name__)
async def repository_error_handler(request: Request, exc: RepositoryError):
log = logger.warning if exc.client_visible else logger.error
log(f"[FastAPI] {request.method} {request.url.path} failed: {exc}")
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.code, "detail": exc.client_message},
)
async def validation_error_handler(request: Request, exc: RequestValidationError):
errors = exc.errors()
for err in errors:
original = (err.get("ctx") or {}).get("error")
if isinstance(original, RepositoryError):
return JSONResponse(
status_code=original.status_code,
content={"code": original.code, "detail": original.client_message},
)
# Problems in headers, path or query are the calling service's fault, not the user's
if any(err["loc"][0] != "body" for err in errors):
logger.error(f"[FastAPI] Invalid request from a calling service: {[(e['loc'], e['msg']) for e in errors]}")
return JSONResponse(status_code=400, content={"code": "bad_request", "detail": "The request could not be processed"})
messages = []
for err in errors:
field = ".".join(str(p) for p in err["loc"] if p != "body")
msg = err["msg"].removeprefix("Value error, ")
messages.append(f"{field}: {msg}" if field else msg)
return JSONResponse(status_code=422, content={"code": "validation_error", "detail": "; ".join(messages)})
async def prometheus_middleware(request: Request, call_next):
logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}")
status = 500
try:
response = await call_next(request)
status = response.status_code
logger.debug(f"[FastAPI] Request processed with status {status}")
except Exception as e:
logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True)
raise
finally:
REQUEST_COUNTER.labels(request.method, request.url.path, status).inc()
logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]")
return response
def register_handlers(app: FastAPI) -> None:
"""Attach the error handlers and the metrics middleware to the app."""
app.add_exception_handler(RepositoryError, repository_error_handler)
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_middleware(BaseHTTPMiddleware, dispatch=prometheus_middleware)
+214 -86
View File
@@ -3,20 +3,33 @@ import os
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
import uvicorn import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, Response from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
from service_auth import UserId
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, get_db, start_healthcheck_thread
from dockerhub_api import find_package_version_with_tag as find_package_version_with_tag_dockerhub from handlers import register_handlers
from github_api import find_package_version_with_tag as find_package_version_with_tag_github from metrics_server import metrics_api
from metrics_server import REQUEST_COUNTER, metrics_api from models import RepositoryType, RepositoryUpdate
from repositories import (
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 secret_manager import cleanup_secret_files
from send_notifications import NotificationClient from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError
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("[FastAPI] Starting application...") logger.info("[FastAPI] Starting application...")
@@ -51,103 +64,218 @@ api = FastAPI(
dependencies=[Depends(auth.authenticate)], dependencies=[Depends(auth.authenticate)],
) )
@api.middleware("http") register_handlers(api)
async def prometheus_middleware(request: Request, call_next):
logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}")
status = 500
try:
response = await call_next(request)
status = response.status_code
logger.debug(f"[FastAPI] Request processed with status {status}")
except Exception as e:
logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True)
raise
finally:
REQUEST_COUNTER.labels(request.method, request.url.path, status).inc()
logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]")
return response
user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
@scheduler_router.post("/github")
def handle_suwayomi( @scheduler_router.post("/update/{repository_type}")
def update_watched_repositories(
request: Request, request: Request,
db = Depends(get_db) 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: try:
logger.info("[FastAPI] Suwayomi handler invoked")
online_version = find_package_version_with_tag_github("Suwayomi", "tachidesk", "stable") logger.info(f"[FastAPI] New version detected for org={org}, repo={repo}, tag={tag}")
logger.debug(f"[FastAPI] Fetched latest Suwayomi version from GitHub: {online_version}")
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute("SELECT latest_version FROM docker_repositories WHERE app='suwayomi'") cursor.execute(
local_state = cursor.fetchone() "SELECT user_id, custom_name "
local_version = str(local_state[0]) if local_state and local_state[0] is not None else None "FROM watched_repositories_subscriptions "
logger.debug(f"[FastAPI] Comparing versions: local='{local_version}' vs online='{online_version}'") "WHERE repository_id = %s AND active = 1",
(repository_id,),
if local_version != online_version:
logger.debug("[FastAPI] Version mismatch detected. Updating database.")
cursor.execute ("UPDATE docker_repositories SET latest_version=%s WHERE app='suwayomi'",
(online_version,))
db.commit()
logger.info("[FastAPI] New Suwayomi version recorded in database")
request.app.state.notifier.send(
title="New Suwayomi version has been found",
body=f"Version {online_version}",
category="utility",
user_id=1,
) )
logger.debug("[FastAPI] Notification sent for Suwayomi update") subscribers = cursor.fetchall()
return Response(status_code=200)
logger.debug("[FastAPI] No new Suwayomi version found") sent = 0
return Response(status_code=204) failed = 0
for user_id, custom_name in subscribers:
except Exception as e:
logger.error(f"[FastAPI] Error in Suwayomi handler: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e)) from e
@scheduler_router.post("/dockerhub")
def handle_pihole(
request: Request,
db = Depends(get_db)
):
try: try:
logger.info("[FastAPI] Pi-hole handler invoked") request.app.state.notifier.send(
online_version = find_package_version_with_tag_dockerhub("pihole/pihole", "latest") title=custom_name or f"{org}/{repo}",
logger.debug(f"[FastAPI] Fetched latest Pi-hole version from Docker Hub: {online_version}") 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: with db.cursor() as cursor:
cursor.execute("SELECT latest_version FROM docker_repositories WHERE app='pihole'") cursor.execute(
local_state = cursor.fetchone() "UPDATE watched_repositories SET latest_entry = %s WHERE id = %s",
local_version = str(local_state[0]) if local_state and local_state[0] is not None else None (online_version, repository_id),
logger.debug(f"[FastAPI] Comparing versions: local='{local_version}' vs online='{online_version}'")
if local_version != online_version:
logger.debug("[FastAPI] Version mismatch detected. Updating database.")
cursor.execute ("UPDATE docker_repositories SET latest_version=%s WHERE app='pihole'",
(online_version,))
db.commit()
logger.info("[FastAPI] New Pi-hole version recorded in database")
request.app.state.notifier.send(
title="New Pi-hole version has been found",
body=f"Version {online_version}",
category="utility",
user_id=1,
) )
logger.debug("[FastAPI] Notification sent for Pi-hole update") db.commit()
return Response(status_code=200) logger.debug(f"[FastAPI] Notification sent and database updated for repository ID {repository_id}")
results["new"] += 1
logger.debug("[FastAPI] No new Pi-hole version found")
return Response(status_code=204)
except Exception as e: except Exception as e:
logger.error(f"[FastAPI] Error in Pi-hole handler: {e}", exc_info=True) db.rollback()
raise HTTPException(status_code=500, detail=str(e)) from e 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(user_router)
api.include_router(scheduler_router) api.include_router(scheduler_router)
+52
View File
@@ -0,0 +1,52 @@
import string
from enum import StrEnum
from pydantic import BaseModel, Field, field_validator
from repository_errors import InvalidOrganisationError, InvalidRepositoryError, InvalidTagError
_REPO_CHARS = frozenset(string.ascii_letters + string.digits + "_.-")
_MAX_ORG = 39
_MAX_REPO = 255
_MAX_TAG = 128
class RepositoryType(StrEnum):
GITHUB = "github"
DOCKERHUB = "dockerhub"
class RepositoryUpdate(BaseModel):
type: RepositoryType
org: str = Field(max_length=_MAX_ORG)
repo: str = Field(max_length=_MAX_REPO)
tag: str = Field(max_length=_MAX_TAG)
custom_name: str | None = Field(default=None, max_length=255)
@field_validator("org")
@classmethod
def _validate_org(cls, v: str) -> str:
v = v.strip()
if v and _REPO_CHARS.issuperset(v) and v[0] not in ".-":
return v
raise InvalidOrganisationError(v)
@field_validator("repo")
@classmethod
def _validate_repo(cls, v: str) -> str:
v = v.strip()
if v and _REPO_CHARS.issuperset(v) and v[0] not in ".-":
return v
raise InvalidRepositoryError(v)
@field_validator("tag")
@classmethod
def _validate_tag(cls, v: str) -> str:
v = v.strip()
if v and _REPO_CHARS.issuperset(v) and v[0] not in ".-":
return v
raise InvalidTagError(v)
@field_validator("custom_name")
@classmethod
def _blank_name_is_none(cls, v: str | None) -> str | None:
return (v or "").strip() or None
+123
View File
@@ -0,0 +1,123 @@
from typing import assert_never
from dockerhub_api import (
find_package_version_with_tag as find_package_version_with_tag_dockerhub,
)
from dockerhub_api import (
validate_repository_combination_and_return_latest_tag as validate_repo_dockerhub,
)
from github_api import (
find_package_version_with_tag as find_package_version_with_tag_github,
)
from github_api import (
validate_repository_combination_and_return_latest_tag as validate_repo_github,
)
from models import RepositoryType, RepositoryUpdate
from send_notifications import NotificationError, NotificationUnavailableError
from simple_logger_handler import setup_logger
logger = setup_logger(__name__)
_MAX_MISSING_CHECKS = 4
def fetch_latest(repository_type: RepositoryType, org:str, repo:str, tag:str) -> str:
match repository_type:
case RepositoryType.GITHUB:
return find_package_version_with_tag_github(org, repo, tag)
case RepositoryType.DOCKERHUB:
return find_package_version_with_tag_dockerhub(org, repo, tag)
case _:
assert_never(repository_type)
def validate_repository_combination_and_return_latest_tag(repository: RepositoryUpdate) -> str:
match repository.type:
case RepositoryType.DOCKERHUB:
return validate_repo_dockerhub(repository.org, repository.repo, repository.tag)
case RepositoryType.GITHUB:
return validate_repo_github(repository.org, repository.repo, repository.tag)
case _:
assert_never(repository.type)
def unsubscribe_every_subscriber_from_repository(
db, notifier, repository_id: int, org: str, repo: str, tag: str
) -> None:
"""Count a "not found". On the 4th miss in a row, deactivate the repository and tell its subscribers."""
subscribers = []
try:
with db.cursor() as cursor:
cursor.execute(
"SELECT missing_checks FROM watched_repositories WHERE id = %s FOR UPDATE",
(repository_id,),
)
row = cursor.fetchone()
if row is None:
db.rollback()
return
misses = row[0] + 1
if misses < _MAX_MISSING_CHECKS:
cursor.execute(
"UPDATE watched_repositories SET missing_checks = %s WHERE id = %s",
(misses, repository_id),
)
else:
cursor.execute(
"SELECT user_id, custom_name FROM watched_repositories_subscriptions "
"WHERE repository_id = %s AND active = 1",
(repository_id,),
)
subscribers = cursor.fetchall()
cursor.execute(
"UPDATE watched_repositories_subscriptions SET active = 0 WHERE repository_id = %s",
(repository_id,),
)
cursor.execute(
"UPDATE watched_repositories SET active = 0, missing_checks = 0 WHERE id = %s",
(repository_id,),
)
db.commit()
except Exception:
db.rollback()
logger.exception(f"[FastAPI] Failed to record a missing check for repository ID {repository_id}")
return
if misses < _MAX_MISSING_CHECKS:
logger.warning(
f"[FastAPI] {org}/{repo}:{tag} not found ({misses}/{_MAX_MISSING_CHECKS}), "
f"checking again on the next run"
)
return
logger.warning(
f"[FastAPI] {org}/{repo}:{tag} not found {_MAX_MISSING_CHECKS} times in a row, deactivated "
f"repository ID {repository_id} and {len(subscribers)} subscription(s)"
)
for user_id, custom_name in subscribers:
try:
notifier.send(
title=custom_name or f"{org}/{repo}",
body=f"{org}/{repo}:{tag} no longer exists, so you were unsubscribed from it",
category="utility",
user_id=user_id,
)
except NotificationUnavailableError:
logger.warning("[FastAPI] Notifications backend unavailable, not informing the remaining users")
break
except NotificationError as e:
logger.warning(
f"[FastAPI] Could not inform user_id={user_id} about the removal "
f"status={getattr(e, 'status_code', None)}"
)
def reset_missing_checks(db, repository_id: int) -> None:
try:
with db.cursor() as cursor:
cursor.execute(
"UPDATE watched_repositories SET missing_checks = 0 WHERE id = %s",
(repository_id,),
)
db.commit()
except Exception:
db.rollback()
logger.exception(f"[FastAPI] Failed to reset the missing counter for repository ID {repository_id}")