refactor(plex): convert service-layer db.query() to 2.0 select() (wave 1)

Mechanical 1.x->2.0 conversion of the self-contained plex modules, behaviour-
equivalent (the characterization net + existing plex tests stay green, 39
passed): app/plex/{sync(15),watch_sync(9),people(1),imgcache(4)} and
app/routes/plex/{admin(4),playback(8),_shared(1)}. Query builders become
select(); .first()/.all()->execute().scalars(); a bulk .update() ->
execute(update()); a single-column .first() -> db.scalar().

browse.py (61) + rollup.agg_subquery (the atomic unit with unified_library)
remain for wave 2.
This commit is contained in:
2026-08-16 05:52:40 +02:00
parent ecbf5737b6
commit 59d50615c4
7 changed files with 174 additions and 67 deletions
+13 -4
View File
@@ -9,6 +9,7 @@ The cache directory (``cache.CACHE_DIR``) is shared with the subtitle cache (see
"""
from fastapi import Response
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import PlexCollection, PlexItem, PlexSeason, PlexShow
@@ -24,16 +25,24 @@ SIZES = {"thumb": (400, 600), "art": (1280, 720)}
def key_for(db: Session, rating_key: str, variant: str) -> str | None:
"""The stored Plex image path for a known rating_key (item/show/season). Only mirrored keys
are proxyable — this is not an open image proxy."""
it = db.query(PlexItem).filter_by(rating_key=rating_key).first()
it = db.execute(
select(PlexItem).where(PlexItem.rating_key == rating_key)
).scalars().first()
if it is not None:
return it.art_key if variant == "art" else it.thumb_key
sh = db.query(PlexShow).filter_by(rating_key=rating_key).first()
sh = db.execute(
select(PlexShow).where(PlexShow.rating_key == rating_key)
).scalars().first()
if sh is not None:
return sh.art_key if variant == "art" else sh.thumb_key
se = db.query(PlexSeason).filter_by(rating_key=rating_key).first()
se = db.execute(
select(PlexSeason).where(PlexSeason.rating_key == rating_key)
).scalars().first()
if se is not None:
return se.thumb_key
col = db.query(PlexCollection).filter_by(rating_key=rating_key).first()
col = db.execute(
select(PlexCollection).where(PlexCollection.rating_key == rating_key)
).scalars().first()
if col is not None:
return col.thumb_key
return None
+5 -2
View File
@@ -5,7 +5,7 @@ re-sync (plex.sync), so the two paths never diverge."""
from urllib.parse import quote
from sqlalchemy import text
from sqlalchemy import select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session
@@ -55,7 +55,10 @@ def enrich_row_people(db: Session, row, cast: list[dict]) -> bool:
dirty = True
# Headshot cache: upsert name -> proxied thumb (a later fetch fills a prior NULL; names without a
# whitelisted photo still get a row so a card can show a placeholder).
existing = {p.name: p for p in db.query(PlexPerson).filter(PlexPerson.name.in_(names))}
existing = {
p.name: p
for p in db.execute(select(PlexPerson).where(PlexPerson.name.in_(names))).scalars()
}
missing: dict[str, str | None] = {}
for c in cast:
nm, th = c.get("name"), c.get("thumb")
+65 -23
View File
@@ -12,6 +12,7 @@ Cast + intro/credit markers are fetched lazily on the item-detail/player path (P
import logging
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app import sysconfig
@@ -165,20 +166,20 @@ def _enrich_people_batch(db: Session, plex: PlexClient, limit: int = _ENRICH_BAT
shared enrich_row_people). Oldest-id first, movies before shows. Stamps people_enriched_at only on
a SUCCESSFUL fetch, so a transient per-item failure retries next run rather than being lost.
Returns how many items were enriched this run."""
movies = (
db.query(PlexItem)
.filter(PlexItem.kind == "movie", PlexItem.people_enriched_at.is_(None))
movies = db.execute(
select(PlexItem)
.where(PlexItem.kind == "movie", PlexItem.people_enriched_at.is_(None))
.order_by(PlexItem.id)
.limit(limit)
.all()
)
).scalars().all()
remaining = limit - len(movies)
shows = (
db.query(PlexShow)
.filter(PlexShow.people_enriched_at.is_(None))
.order_by(PlexShow.id)
.limit(remaining)
.all()
db.execute(
select(PlexShow)
.where(PlexShow.people_enriched_at.is_(None))
.order_by(PlexShow.id)
.limit(remaining)
).scalars().all()
if remaining > 0
else []
)
@@ -198,7 +199,9 @@ def _enrich_people_batch(db: Session, plex: PlexClient, limit: int = _ENRICH_BAT
def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary:
lib = db.query(PlexLibrary).filter_by(plex_key=key).first()
lib = db.execute(
select(PlexLibrary).where(PlexLibrary.plex_key == key)
).scalars().first()
if lib is None:
lib = PlexLibrary(plex_key=key)
db.add(lib)
@@ -261,7 +264,12 @@ def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None:
def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
existing = {r.rating_key: r for r in db.query(PlexItem).filter_by(library_id=lib.id, kind="movie")}
existing = {
r.rating_key: r
for r in db.execute(
select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
).scalars()
}
for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE):
rk = str(meta.get("ratingKey") or "")
if not rk:
@@ -276,7 +284,12 @@ def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -
def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
# 1) shows
shows = {r.rating_key: r for r in db.query(PlexShow).filter_by(library_id=lib.id)}
shows = {
r.rating_key: r
for r in db.execute(
select(PlexShow).where(PlexShow.library_id == lib.id)
).scalars()
}
for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW):
rk = str(meta.get("ratingKey") or "")
if not rk:
@@ -299,7 +312,12 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
show_id = {rk: sh.id for rk, sh in shows.items()}
# 2) seasons
seasons = {r.rating_key: r for r in db.query(PlexSeason).join(PlexShow).filter(PlexShow.library_id == lib.id)}
seasons = {
r.rating_key: r
for r in db.execute(
select(PlexSeason).join(PlexShow).where(PlexShow.library_id == lib.id)
).scalars()
}
for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON):
rk = str(meta.get("ratingKey") or "")
sid = show_id.get(str(meta.get("parentRatingKey") or ""))
@@ -318,7 +336,12 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
season_id = {rk: se.id for rk, se in seasons.items()}
# 3) episodes (playable leaves)
episodes = {r.rating_key: r for r in db.query(PlexItem).filter_by(library_id=lib.id, kind="episode")}
episodes = {
r.rating_key: r
for r in db.execute(
select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "episode")
).scalars()
}
for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE):
rk = str(meta.get("ratingKey") or "")
if not rk:
@@ -359,17 +382,27 @@ def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str)
collection's membership on the affected member rows."""
meta = plex.metadata(rk) or {}
col = _upsert_collection(
db, db.query(PlexCollection).filter_by(rating_key=rk).first(), rk, lib.id, meta
db,
db.execute(
select(PlexCollection).where(PlexCollection.rating_key == rk)
).scalars().first(),
rk,
lib.id,
meta,
)
Model = PlexItem if lib.kind == "movie" else PlexShow
new_members = {str(c.get("ratingKey")) for c in plex.collection_children(rk) if c.get("ratingKey")}
current = db.query(Model).filter(Model.collection_keys.contains([rk])).all()
current = db.execute(
select(Model).where(Model.collection_keys.contains([rk]))
).scalars().all()
for row in current: # drop rk from rows that are no longer members
if row.rating_key not in new_members:
row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
to_add = new_members - {r.rating_key for r in current}
if to_add:
for row in db.query(Model).filter(Model.rating_key.in_(to_add)):
for row in db.execute(
select(Model).where(Model.rating_key.in_(to_add))
).scalars():
row.collection_keys = list(dict.fromkeys([*(row.collection_keys or []), rk]))
db.flush()
return col
@@ -378,9 +411,13 @@ def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str)
def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None:
"""Local cleanup after a collection is deleted from Plex: strip rk from members + drop the row."""
Model = PlexItem if lib.kind == "movie" else PlexShow
for row in db.query(Model).filter(Model.collection_keys.contains([rk])):
for row in db.execute(
select(Model).where(Model.collection_keys.contains([rk]))
).scalars():
row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
col = db.query(PlexCollection).filter_by(rating_key=rk).first()
col = db.execute(
select(PlexCollection).where(PlexCollection.rating_key == rk)
).scalars().first()
if col is not None:
db.delete(col)
db.flush()
@@ -391,7 +428,12 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di
written as `collection_keys` on member movies (plex_items) / shows (plex_shows). A full reconcile:
collections gone from Plex are pruned, and each member row's keys are rebuilt from scratch."""
cols = plex.collections(lib.plex_key)
existing = {c.rating_key: c for c in db.query(PlexCollection).filter_by(library_id=lib.id)}
existing = {
c.rating_key: c
for c in db.execute(
select(PlexCollection).where(PlexCollection.library_id == lib.id)
).scalars()
}
seen: set[str] = set()
membership: dict[str, list[str]] = {} # member rating_key -> [collection rating_key, ...]
for meta in cols:
@@ -410,8 +452,8 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di
if rk not in seen:
db.delete(col)
# Rebuild collection_keys on every member row of this library (clear + set).
for row in db.query(PlexItem).filter_by(library_id=lib.id):
for row in db.execute(select(PlexItem).where(PlexItem.library_id == lib.id)).scalars():
row.collection_keys = membership.get(row.rating_key) or None
for row in db.query(PlexShow).filter_by(library_id=lib.id):
for row in db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars():
row.collection_keys = membership.get(row.rating_key) or None
db.flush()
+40 -20
View File
@@ -16,7 +16,7 @@ history + onDeck, last-write-wins) build on this and land in later ships.
import logging
from datetime import datetime, timedelta, timezone
from sqlalchemy import func
from sqlalchemy import func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session
@@ -61,7 +61,10 @@ def _plex_watch_to_state(
def _rk_to_id(db: Session) -> dict[str, int]:
"""The rating_key → PlexItem.id lookup used to resolve a Plex leaf back to its mirrored row."""
return {rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)}
return {
rk: iid
for rk, iid in db.execute(select(PlexItem.rating_key, PlexItem.id)).all()
}
def _scan_plex_states(
@@ -162,7 +165,9 @@ def link_for_push(db: Session, user_id: int) -> PlexLink | None:
"""The active owner watch-push link for this user, or None when push is off / not the owner.
Push is gated on `initial_import_done` so we never push a half-known local state over Plex's
authoritative one before the one-time "Plex is master" import has reconciled them."""
link = db.query(PlexLink).filter_by(user_id=user_id).first()
link = db.execute(
select(PlexLink).where(PlexLink.user_id == user_id)
).scalars().first()
if link and link.uses_admin and link.sync_enabled and link.initial_import_done:
return link
return None
@@ -217,7 +222,11 @@ def push_state_to_plex(
# Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be
# nothing to flag; that's fine (unscrobble still happened). Skip the flag if the row was
# re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile).
st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first()
st = db.execute(
select(PlexState).where(
PlexState.user_id == user_id, PlexState.item_id == item_id
)
).scalars().first()
if st is not None and _row_matches(st, expect_ts):
st.synced_to_plex = True
db.commit()
@@ -272,9 +281,11 @@ def push_bulk_state_to_plex(user_id: int, changes: list[tuple[int, str, str]]) -
# Flag every successfully-pushed row that still exists (an "unwatched" deletes the row, so
# only the "watched" changes have a row to flag).
if synced_ids:
db.query(PlexState).filter(
PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids)
).update({PlexState.synced_to_plex: True}, synchronize_session=False)
db.execute(
update(PlexState)
.where(PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids))
.values(synced_to_plex=True)
)
db.commit()
@@ -290,11 +301,13 @@ _LWW_TOLERANCE_S = 90
def _active_sync_links(db: Session) -> list[PlexLink]:
"""Every link with two-way sync actually running (owner MVP = one row). Iterated uniformly so
P5b friend links slot in later without touching this loop."""
return (
db.query(PlexLink)
.filter_by(uses_admin=True, sync_enabled=True, initial_import_done=True)
.all()
)
return db.execute(
select(PlexLink).where(
PlexLink.uses_admin.is_(True),
PlexLink.sync_enabled.is_(True),
PlexLink.initial_import_done.is_(True),
)
).scalars().all()
def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int:
@@ -340,7 +353,11 @@ def _pull_apply(
"""Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
"watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
full reconcile's job) — this only pulls in watched/resume."""
st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first()
st = db.execute(
select(PlexState).where(
PlexState.user_id == user_id, PlexState.item_id == item_id
)
).scalars().first()
if _same_state(st, status, pos):
if st is not None and not st.synced_to_plex:
st.synced_to_plex = True # already matches Plex → just settle the flag, no churn
@@ -368,17 +385,18 @@ def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int:
# re-push doesn't need an extra per-item plex.metadata() round-trip just to fill set_timeline's
# duration argument.
info_by_id: dict[int, tuple[str, int | None]] = {
iid: (rk, dur) for iid, rk, dur in db.query(PlexItem.id, PlexItem.rating_key, PlexItem.duration_s)
iid: (rk, dur)
for iid, rk, dur in db.execute(
select(PlexItem.id, PlexItem.rating_key, PlexItem.duration_s)
).all()
}
dirty = (
db.query(PlexState)
.filter(
dirty = db.execute(
select(PlexState).where(
PlexState.user_id == link.user_id,
PlexState.synced_to_plex.is_(False),
PlexState.status != "hidden",
)
.all()
)
).scalars().all()
n = 0
for st in dirty:
info = info_by_id.get(st.item_id)
@@ -477,7 +495,9 @@ def run_plex_watch_reconcile(db: Session) -> dict:
# Siftlode rows Plex no longer has a record for: clear the ones that were mirrored
# (Plex un-watched them); leave hidden + dirty (dirty gets re-pushed just below).
for st in db.query(PlexState).filter_by(user_id=link.user_id).all():
for st in db.execute(
select(PlexState).where(PlexState.user_id == link.user_id)
).scalars().all():
if st.item_id in plex_states or st.status == "hidden":
continue
if st.synced_to_plex:
+4 -1
View File
@@ -3,6 +3,7 @@
import logging
from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import (
@@ -13,7 +14,9 @@ log = logging.getLogger("siftlode.plex")
def item_or_404(db: Session, rating_key: str) -> PlexItem:
it = db.query(PlexItem).filter_by(rating_key=str(rating_key)).first()
it = db.execute(
select(PlexItem).where(PlexItem.rating_key == str(rating_key))
).scalars().first()
if it is None:
raise HTTPException(status_code=404, detail="Unknown Plex item")
return it
+13 -6
View File
@@ -4,6 +4,7 @@ from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from app import sysconfig
@@ -65,9 +66,9 @@ def _sample_plex_path(
"""A real Plex-side media file path to validate the mount + path-map against. Prefers the
already-mirrored catalog (cheap); before the first sync, pulls one movie leaf from an enabled
section straight from Plex."""
row = db.query(PlexItem.file_path).filter(PlexItem.file_path.isnot(None)).first()
if row and row[0]:
return row[0]
fp = db.scalar(select(PlexItem.file_path).where(PlexItem.file_path.isnot(None)))
if fp:
return fp
wanted = plex_sync._enabled_section_keys(db)
for s in sections:
if s.get("type") != "movie":
@@ -117,7 +118,9 @@ def watch_link_status(
user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
"""This user's Plex watch-sync link status (drives the Settings toggle)."""
link = db.query(PlexLink).filter_by(user_id=user.id).first()
link = db.execute(
select(PlexLink).where(PlexLink.user_id == user.id)
).scalars().first()
return _link_dict(link)
@@ -137,7 +140,9 @@ def watch_link_set(
if not sysconfig.get_bool(db, "plex_enabled"):
raise HTTPException(status_code=409, detail="The Plex module is disabled.")
enabled = body.enabled
link = db.query(PlexLink).filter_by(user_id=user.id).first()
link = db.execute(
select(PlexLink).where(PlexLink.user_id == user.id)
).scalars().first()
if link is None:
link = PlexLink(
user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc)
@@ -158,7 +163,9 @@ def watch_import_now(
) -> dict:
"""Admin: (re-)run the "Plex is master" import for this account — Plex re-wins on the
intersection; Siftlode-only states are preserved. Idempotent."""
link = db.query(PlexLink).filter_by(user_id=user.id).first()
link = db.execute(
select(PlexLink).where(PlexLink.user_id == user.id)
).scalars().first()
if link is None or not link.sync_enabled:
raise HTTPException(status_code=409, detail="Enable Plex watch-sync first.")
result = plex_watch.import_owner_watch_state(db, link)
+34 -11
View File
@@ -16,6 +16,7 @@ from fastapi import (
)
from fastapi.responses import FileResponse
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.auth import current_user, resolved_user_id
@@ -207,7 +208,11 @@ def _push_watch(
return
# Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit
# lands before it settles synced_to_plex (which would otherwise strand that newer value).
st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first()
st = db.execute(
select(PlexState).where(
PlexState.user_id == user_id, PlexState.item_id == item_id
)
).scalars().first()
background.add_task(
plex_watch.push_state_to_plex,
user_id,
@@ -246,7 +251,11 @@ def item_progress(
duration = int(body.duration_seconds)
final = body.final
now = datetime.now(timezone.utc)
st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
st = db.execute(
select(PlexState).where(
PlexState.user_id == user.id, PlexState.item_id == it.id
)
).scalars().first()
was_watched = st is not None and st.status == "watched"
near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
@@ -316,7 +325,11 @@ def item_state(
status = body.status
if status not in ("new", "watched", "hidden"):
raise HTTPException(status_code=400, detail="Invalid status")
st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
st = db.execute(
select(PlexState).where(
PlexState.user_id == user.id, PlexState.item_id == it.id
)
).scalars().first()
prev_status = st.status if st is not None else "new"
if status == "new":
if st is not None:
@@ -351,10 +364,12 @@ def _bulk_state(
now = datetime.now(timezone.utc)
existing = {
s.item_id: s
for s in db.query(PlexState).filter(
PlexState.user_id == user_id,
PlexState.item_id.in_([e.id for e in eps] or [0]),
)
for s in db.execute(
select(PlexState).where(
PlexState.user_id == user_id,
PlexState.item_id.in_([e.id for e in eps] or [0]),
)
).scalars()
}
pushable = plex_watch.link_for_push(db, user_id) is not None
to_push: list[tuple[int, str, str]] = []
@@ -397,11 +412,15 @@ def show_state(
db: Session = Depends(get_db),
) -> dict:
"""Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex."""
sh = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first()
sh = db.execute(
select(PlexShow).where(PlexShow.rating_key == str(rating_key))
).scalars().first()
if sh is None:
raise HTTPException(status_code=404, detail="Unknown Plex show")
watched = body.watched
eps = db.query(PlexItem).filter_by(show_id=sh.id, kind="episode").all()
eps = db.execute(
select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode")
).scalars().all()
changed = _bulk_state(background, db, user.id, eps, watched)
return {"changed": changed, "watched": watched}
@@ -415,11 +434,15 @@ def season_state(
db: Session = Depends(get_db),
) -> dict:
"""Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex."""
se = db.query(PlexSeason).filter_by(rating_key=str(rating_key)).first()
se = db.execute(
select(PlexSeason).where(PlexSeason.rating_key == str(rating_key))
).scalars().first()
if se is None:
raise HTTPException(status_code=404, detail="Unknown Plex season")
watched = body.watched
eps = db.query(PlexItem).filter_by(season_id=se.id, kind="episode").all()
eps = db.execute(
select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode")
).scalars().all()
changed = _bulk_state(background, db, user.id, eps, watched)
return {"changed": changed, "watched": watched}