chore(scheduler): dedup token-user query + drop pointless rss lambda

- Extract runner.token_users(db): the "all users with a stored refresh token"
  query was spelled 3× (get_service_user, run_subscription_resync,
  sync_all_playlists). get_service_user now reuses it ([0]); sync_all_playlists
  imports it from runner (no cycle — runner doesn't import playlists); the now-
  unused OAuthToken import is dropped there.
- scheduler.py: _job("rss_poll", lambda db: run_rss_poll(db)) → _job("rss_poll",
  run_rss_poll) — the lambda was dead indirection; all sibling jobs pass the
  callable directly and _job calls fn(db).

Behavior-neutral. ruff clean, localdev boots, re-review clean.
This commit is contained in:
npeter83 2026-07-11 19:44:53 +02:00
parent 4e68bdf920
commit a6ffcf9577
3 changed files with 14 additions and 23 deletions

View file

@ -209,7 +209,7 @@ def scheduler_snapshot() -> dict:
def _rss_job() -> None:
_job("rss_poll", lambda db: run_rss_poll(db))
_job("rss_poll", run_rss_poll)
def _enrich_job() -> None:

View file

@ -14,7 +14,8 @@ from sqlalchemy.orm import Session
from app import progress, quota
from app.auth import has_read_scope
from app.models import Channel, OAuthToken, Playlist, PlaylistItem, User, Video
from app.models import Channel, Playlist, PlaylistItem, User, Video
from app.sync.runner import token_users
from app.sync.videos import apply_video_details, parse_dt
from app.youtube.client import YouTubeClient, YouTubeError
@ -370,13 +371,7 @@ def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
def sync_all_playlists(db: Session) -> dict:
"""Scheduler entry point: mirror playlists for every user with a read scope + token."""
users = (
db.execute(
select(User).join(OAuthToken).where(OAuthToken.refresh_token_enc.is_not(None))
)
.scalars()
.all()
)
users = token_users(db)
total = 0
for i, user in enumerate(users, 1):
if not has_read_scope(user):

View file

@ -31,17 +31,21 @@ from app.youtube.rss import fetch_channel_feed
RSS_POLL_WORKERS = 16
def get_service_user(db: Session) -> User | None:
return (
def token_users(db: Session) -> list[User]:
"""Every user with a stored YouTube refresh token (the read-scope service accounts), by id."""
return list(
db.execute(
select(User)
.join(OAuthToken)
.where(OAuthToken.refresh_token_enc.is_not(None))
.order_by(User.id)
).scalars()
)
.scalars()
.first()
)
def get_service_user(db: Session) -> User | None:
users = token_users(db)
return users[0] if users else None
def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
@ -102,15 +106,7 @@ def run_shorts(db: Session) -> dict:
def run_subscription_resync(db: Session) -> dict:
"""Re-import every user's subscriptions so unsubscribes and new subscriptions on
YouTube are reflected automatically."""
users = (
db.execute(
select(User)
.join(OAuthToken)
.where(OAuthToken.refresh_token_enc.is_not(None))
)
.scalars()
.all()
)
users = token_users(db)
for i, user in enumerate(users, 1):
try:
import_subscriptions(db, user)