diff --git a/backend/app/plex/imgcache.py b/backend/app/plex/imgcache.py index 2dbeb2a3..215e18ea 100644 --- a/backend/app/plex/imgcache.py +++ b/backend/app/plex/imgcache.py @@ -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 diff --git a/backend/app/plex/people.py b/backend/app/plex/people.py index 7f5652c7..758ce276 100644 --- a/backend/app/plex/people.py +++ b/backend/app/plex/people.py @@ -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") diff --git a/backend/app/plex/sync.py b/backend/app/plex/sync.py index f1b9e82a..cdfa0711 100644 --- a/backend/app/plex/sync.py +++ b/backend/app/plex/sync.py @@ -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() diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 6e1c3d7e..a72d5b01 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -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: diff --git a/backend/app/routes/plex/_shared.py b/backend/app/routes/plex/_shared.py index ecd7f8c2..a35e216e 100644 --- a/backend/app/routes/plex/_shared.py +++ b/backend/app/routes/plex/_shared.py @@ -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 diff --git a/backend/app/routes/plex/admin.py b/backend/app/routes/plex/admin.py index 5c5ddee8..4681542b 100644 --- a/backend/app/routes/plex/admin.py +++ b/backend/app/routes/plex/admin.py @@ -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) diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 42495705..fed33581 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -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}