feat(scheduler): admin-editable job intervals (DB-backed, live reschedule)

Add scheduler_settings (per-job interval override, migration 0015) and a
PATCH /api/admin/scheduler/jobs/{id} that persists the override and live-
reschedules the running APScheduler job. Intervals load from the DB (env
defaults as fallback); the snapshot reports the live trigger interval.
This commit is contained in:
npeter83 2026-06-16 15:58:23 +02:00
parent bad7bba3f9
commit 79d645c2e2
4 changed files with 126 additions and 41 deletions

View file

@ -0,0 +1,30 @@
"""scheduler job interval overrides
Revision ID: 0015_scheduler_settings
Revises: 0014_demo_account
Create Date: 2026-06-16
Adds scheduler_settings: a per-job run-interval override (minutes), editable from the admin
Scheduler dashboard. Absent row = use the env/config default, so this is purely additive.
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "0015_scheduler_settings"
down_revision: Union[str, None] = "0014_demo_account"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"scheduler_settings",
sa.Column("job_id", sa.String(length=40), primary_key=True),
sa.Column("interval_minutes", sa.Integer(), nullable=False),
)
def downgrade() -> None:
op.drop_table("scheduler_settings")

View file

@ -325,6 +325,17 @@ class AppState(Base):
) )
class SchedulerSetting(Base):
"""Admin override for a scheduler job's run interval (minutes). One row per job id;
absent = use the env/config default. DB-backed so it's editable from the admin UI at
runtime without a redeploy (env stays a fallback)."""
__tablename__ = "scheduler_settings"
job_id: Mapped[str] = mapped_column(String(40), primary_key=True)
interval_minutes: Mapped[int] = mapped_column(Integer)
class Playlist(Base): class Playlist(Base):
"""A per-user named, ordered collection of videos from the shared catalog. """A per-user named, ordered collection of videos from the shared catalog.

View file

@ -1,7 +1,7 @@
"""Admin Scheduler dashboard: a live view of what the background scheduler is doing — """Admin Scheduler dashboard: a live view of what the background scheduler is doing —
per-job run activity (from the in-process scheduler), the work still queued (DB-derived), per-job run activity (from the in-process scheduler), the work still queued (DB-derived),
and the shared quota picture.""" and the shared quota picture."""
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, func, or_, select from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
@ -10,12 +10,36 @@ from app.config import settings
from app.db import get_db from app.db import get_db
from app.models import Channel, Subscription, User, Video from app.models import Channel, Subscription, User, Video
from app.routes.admin import admin_user from app.routes.admin import admin_user
from app.scheduler import scheduler_snapshot from app.scheduler import JOB_INTERVALS, MAX_INTERVAL, MIN_INTERVAL, apply_interval, scheduler_snapshot
from app.sync.runner import estimate_deep_backfill from app.sync.runner import estimate_deep_backfill
router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"]) router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"])
@router.patch("/jobs/{job_id}")
def set_job_interval(
job_id: str,
payload: dict,
_: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
"""Change how often a scheduler job runs (minutes). Persisted as an override and applied
to the live scheduler immediately."""
if job_id not in JOB_INTERVALS:
raise HTTPException(status_code=404, detail="Unknown job")
try:
minutes = int(payload.get("interval_minutes"))
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="interval_minutes must be a number")
if not (MIN_INTERVAL <= minutes <= MAX_INTERVAL):
raise HTTPException(
status_code=400,
detail=f"interval must be between {MIN_INTERVAL} and {MAX_INTERVAL} minutes",
)
applied = apply_interval(db, job_id, minutes)
return {"id": job_id, "interval_minutes": applied}
@router.get("") @router.get("")
def get_scheduler( def get_scheduler(
_: User = Depends(admin_user), db: Session = Depends(get_db) _: User = Depends(admin_user), db: Session = Depends(get_db)

View file

@ -5,10 +5,12 @@ import threading
from datetime import datetime, timezone from datetime import datetime, timezone
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import select
from app import quota from app import quota
from app.config import settings from app.config import settings
from app.db import SessionLocal from app.db import SessionLocal
from app.models import SchedulerSetting
from app.state import is_sync_paused from app.state import is_sync_paused
from app.sync.autotag import run_autotag_all from app.sync.autotag import run_autotag_all
from app.sync.playlists import sync_all_playlists from app.sync.playlists import sync_all_playlists
@ -32,7 +34,8 @@ _scheduler: BackgroundScheduler | None = None
_activity: dict[str, dict] = {} _activity: dict[str, dict] = {}
_activity_lock = threading.Lock() _activity_lock = threading.Lock()
# Each interval job's configured period, surfaced so the dashboard can show "every N min". # Each interval job's env/config default period (minutes). The admin can override any of
# these at runtime (stored in scheduler_settings, applied live); see load_intervals.
JOB_INTERVALS: dict[str, int] = { JOB_INTERVALS: dict[str, int] = {
"rss_poll": settings.rss_poll_minutes, "rss_poll": settings.rss_poll_minutes,
"enrich": settings.enrich_interval_minutes, "enrich": settings.enrich_interval_minutes,
@ -43,6 +46,38 @@ JOB_INTERVALS: dict[str, int] = {
"playlist_sync": settings.playlist_sync_minutes, "playlist_sync": settings.playlist_sync_minutes,
} }
# Sane bounds for an admin-set interval (minutes).
MIN_INTERVAL = 1
MAX_INTERVAL = 1440 # one day
def load_intervals(db) -> dict[str, int]:
"""Effective per-job intervals: the env/config defaults overlaid with any admin overrides
saved in scheduler_settings."""
iv = dict(JOB_INTERVALS)
for row in db.execute(select(SchedulerSetting)).scalars():
if row.job_id in iv and row.interval_minutes:
iv[row.job_id] = row.interval_minutes
return iv
def apply_interval(db, job_id: str, minutes: int) -> int:
"""Persist a job's interval override and reschedule the live job immediately (if the
scheduler runs in this process). Returns the applied value."""
if job_id not in JOB_INTERVALS:
raise KeyError(job_id)
minutes = max(MIN_INTERVAL, min(MAX_INTERVAL, int(minutes)))
row = db.get(SchedulerSetting, job_id)
if row is None:
db.add(SchedulerSetting(job_id=job_id, interval_minutes=minutes))
else:
row.interval_minutes = minutes
db.commit()
if _scheduler is not None and _scheduler.get_job(job_id) is not None:
_scheduler.reschedule_job(job_id, trigger="interval", minutes=minutes)
logger.info("rescheduled job %s to every %s min", job_id, minutes)
return minutes
def _now_iso() -> str: def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat() return datetime.now(timezone.utc).isoformat()
@ -90,19 +125,23 @@ def scheduler_snapshot() -> dict:
runs no scheduler), so the UI can say so while still showing DB-derived queue/quota.""" runs no scheduler), so the UI can say so while still showing DB-derived queue/quota."""
running_here = _scheduler is not None running_here = _scheduler is not None
next_runs: dict[str, str | None] = {} next_runs: dict[str, str | None] = {}
live_intervals: dict[str, int] = {}
if _scheduler is not None: if _scheduler is not None:
for job in _scheduler.get_jobs(): for job in _scheduler.get_jobs():
nr = getattr(job, "next_run_time", None) nr = getattr(job, "next_run_time", None)
next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None
iv = getattr(job.trigger, "interval", None)
if iv is not None:
live_intervals[job.id] = round(iv.total_seconds() / 60)
with _activity_lock: with _activity_lock:
acts = {k: dict(v) for k, v in _activity.items()} acts = {k: dict(v) for k, v in _activity.items()}
jobs = [] jobs = []
for job_id, interval in JOB_INTERVALS.items(): for job_id, default_interval in JOB_INTERVALS.items():
a = acts.get(job_id, {}) a = acts.get(job_id, {})
jobs.append( jobs.append(
{ {
"id": job_id, "id": job_id,
"interval_minutes": interval, "interval_minutes": live_intervals.get(job_id, default_interval),
"next_run": next_runs.get(job_id), "next_run": next_runs.get(job_id),
"running": a.get("running", False), "running": a.get("running", False),
"status": a.get("status"), "status": a.get("status"),
@ -166,43 +205,24 @@ def start_scheduler() -> None:
global _scheduler global _scheduler
if not settings.scheduler_enabled or _scheduler is not None: if not settings.scheduler_enabled or _scheduler is not None:
return return
# Effective intervals = env defaults overlaid with any admin overrides from the DB.
db = SessionLocal()
try:
iv = load_intervals(db)
finally:
db.close()
fns = {
"rss_poll": _rss_job,
"enrich": _enrich_job,
"backfill": _backfill_job,
"autotag": _autotag_job,
"shorts": _shorts_job,
"subscriptions": _subscriptions_job,
"playlist_sync": _playlist_sync_job,
}
scheduler = BackgroundScheduler(timezone="UTC") scheduler = BackgroundScheduler(timezone="UTC")
scheduler.add_job( for job_id, fn in fns.items():
_rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll" scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
)
scheduler.add_job(
_enrich_job, "interval", minutes=settings.enrich_interval_minutes, id="enrich"
)
scheduler.add_job(
_backfill_job,
"interval",
minutes=settings.backfill_interval_minutes,
id="backfill",
)
scheduler.add_job(
_autotag_job,
"interval",
minutes=settings.autotag_interval_minutes,
id="autotag",
)
scheduler.add_job(
_shorts_job,
"interval",
minutes=settings.shorts_probe_interval_minutes,
id="shorts",
)
scheduler.add_job(
_subscriptions_job,
"interval",
minutes=settings.subscriptions_resync_minutes,
id="subscriptions",
)
scheduler.add_job(
_playlist_sync_job,
"interval",
minutes=settings.playlist_sync_minutes,
id="playlist_sync",
)
scheduler.start() scheduler.start()
_scheduler = scheduler _scheduler = scheduler
logger.info("scheduler started") logger.info("scheduler started")