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 b71d07c098
commit 84aebe16c7
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):
"""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 —
per-job run activity (from the in-process scheduler), the work still queued (DB-derived),
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.orm import Session
@ -10,12 +10,36 @@ from app.config import settings
from app.db import get_db
from app.models import Channel, Subscription, User, Video
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
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("")
def get_scheduler(
_: User = Depends(admin_user), db: Session = Depends(get_db)

View file

@ -5,10 +5,12 @@ import threading
from datetime import datetime, timezone
from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import select
from app import quota
from app.config import settings
from app.db import SessionLocal
from app.models import SchedulerSetting
from app.state import is_sync_paused
from app.sync.autotag import run_autotag_all
from app.sync.playlists import sync_all_playlists
@ -32,7 +34,8 @@ _scheduler: BackgroundScheduler | None = None
_activity: dict[str, dict] = {}
_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] = {
"rss_poll": settings.rss_poll_minutes,
"enrich": settings.enrich_interval_minutes,
@ -43,6 +46,38 @@ JOB_INTERVALS: dict[str, int] = {
"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:
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."""
running_here = _scheduler is not None
next_runs: dict[str, str | None] = {}
live_intervals: dict[str, int] = {}
if _scheduler is not None:
for job in _scheduler.get_jobs():
nr = getattr(job, "next_run_time", 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:
acts = {k: dict(v) for k, v in _activity.items()}
jobs = []
for job_id, interval in JOB_INTERVALS.items():
for job_id, default_interval in JOB_INTERVALS.items():
a = acts.get(job_id, {})
jobs.append(
{
"id": job_id,
"interval_minutes": interval,
"interval_minutes": live_intervals.get(job_id, default_interval),
"next_run": next_runs.get(job_id),
"running": a.get("running", False),
"status": a.get("status"),
@ -166,43 +205,24 @@ def start_scheduler() -> None:
global _scheduler
if not settings.scheduler_enabled or _scheduler is not None:
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.add_job(
_rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll"
)
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",
)
for job_id, fn in fns.items():
scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
scheduler.start()
_scheduler = scheduler
logger.info("scheduler started")