refactor(api): shared serializer + ownership + write-scope helpers (R6 S1a)

Establish the R6 API-contract infra before the S1b payload:dict migration:
- iso(dt) in utils.py replaces the ~16 hand-rolled `x.isoformat() if x else None`
  serializer idioms across the routes.
- get_or_404 / owned_or_404 in routes/_common.py — one 404 + existence-leak policy;
  plex's _own_playlist_or_404 and saved_views' _own_view now delegate to it.
- require_write_scope dependency in auth.py collapses the inline has_write_scope→403
  checks (channels subscribe/unsubscribe, playlists push/push-plan) that had drifted
  into three different messages into one; the conditional delete_playlist guard stays
  inline (its message was already canonical).

No response-shape change (iso == isoformat for a present value). Gate: ruff clean,
pytest 163 green (+ test_utils for iso).

backend/app/auth.py

@@ -1010,6 +1010,19 @@ def require_human(user: User = Depends(current_user)) -> User:
     return user
 
 
+def require_write_scope(user: User = Depends(require_human)) -> User:
+    """Dependency: require the YouTube write scope (subscribe/unsubscribe, playlist export/edit).
+    Single source of truth for the write gate — replaces the inline `if not has_write_scope(user)`
+    checks that had drifted into three different 403 messages across channels/playlists. Layers on
+    `require_human` because writing to YouTube needs a real, non-demo identity."""
+    if not has_write_scope(user):
+        raise HTTPException(
+            status_code=403,
+            detail="Enable YouTube editing in Settings first.",
+        )
+    return user
+
+
 def admin_user(user: User = Depends(current_user)) -> User:
     """Dependency: require the signed-in user to be an admin. Single source of truth for the
     admin gate — every admin route/action depends on this rather than re-checking the role."""

backend/app/routes/_common.py

@@ -0,0 +1,44 @@
+"""Route-layer shared helpers for the API contract (R6 S1a).
+
+Small, dependency-light guards that several route modules had hand-rolled with subtle
+drift. Keeping them here (imported by the route modules) means the "fetch a row or 404"
+and "fetch a row this user owns or 404" idioms have one definition, one status code, and
+one existence-leak policy.
+"""
+from typing import Any, TypeVar
+
+from fastapi import HTTPException
+from sqlalchemy.orm import Session
+
+from app.db import Base
+from app.models import User
+
+M = TypeVar("M", bound=Base)
+
+
+def get_or_404(db: Session, model: type[M], id_: Any, *, detail: str = "Not found") -> M:
+    """Fetch a row by primary key or raise 404. Replaces the `db.get(Model, id); if None: 404`
+    idiom repeated across routes."""
+    obj = db.get(model, id_)
+    if obj is None:
+        raise HTTPException(status_code=404, detail=detail)
+    return obj
+
+
+def owned_or_404(
+    db: Session,
+    model: type[M],
+    id_: Any,
+    user: User,
+    *,
+    owner_attr: str = "user_id",
+    detail: str = "Not found",
+) -> M:
+    """Fetch a row by primary key that `user` owns, or raise 404. A row owned by someone
+    else 404s (NOT 403) on purpose — a 403 would confirm the id exists to a stranger.
+    Generalizes the near-identical per-route ownership helpers (e.g. plex's
+    `_own_playlist_or_404`, the saved-views owner check)."""
+    obj = get_or_404(db, model, id_, detail=detail)
+    if getattr(obj, owner_attr) != user.id:
+        raise HTTPException(status_code=404, detail=detail)
+    return obj

backend/app/routes/admin.py

@@ -27,7 +27,7 @@ from app.models import (
     Video,
     VideoState,
 )
-from app.utils import valid_email
+from app.utils import iso, valid_email
 
 router = APIRouter(prefix="/api/admin", tags=["admin"])
 
@@ -38,8 +38,8 @@ def _serialize(inv: Invite) -> dict:
         "email": inv.email,
         "status": inv.status,
         "note": inv.note,
-        "requested_at": inv.requested_at.isoformat() if inv.requested_at else None,
-        "decided_at": inv.decided_at.isoformat() if inv.decided_at else None,
+        "requested_at": iso(inv.requested_at),
+        "decided_at": iso(inv.decided_at),
         "decided_by": inv.decided_by,
     }
 
@@ -147,7 +147,7 @@ def _serialize_user(u: User) -> dict:
         "is_demo": u.is_demo,
         "has_password": u.password_hash is not None,
         "has_google": u.google_sub is not None,
-        "created_at": u.created_at.isoformat() if u.created_at else None,
+        "created_at": iso(u.created_at),
     }
 
 
@@ -280,7 +280,7 @@ def _serialize_demo(row: DemoWhitelist) -> dict:
         "email": row.email,
         "note": row.note,
         "added_by": row.added_by,
-        "created_at": row.created_at.isoformat() if row.created_at else None,
+        "created_at": iso(row.created_at),
     }
 
 

backend/app/routes/audit.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
 from app import audit
 from app.audit import AuditAction
 from app.db import get_db
+from app.utils import iso
 from app.models import AuditLog, User
 from app.routes.admin import admin_user
 
@@ -23,7 +24,7 @@ MAX_ROWS = 2000
 def _serialize(row: AuditLog, actors: dict[int, str]) -> dict:
     return {
         "id": row.id,
-        "created_at": row.created_at.isoformat() if row.created_at else None,
+        "created_at": iso(row.created_at),
         # actor_id NULL = the scheduler/background did it (client shows "System"); a set id that
         # can't be resolved would mean a deleted user, but SET NULL makes that NULL — so present.
         "actor": actors.get(row.actor_id) if row.actor_id is not None else None,

backend/app/routes/channels.py

@@ -9,8 +9,9 @@ from sqlalchemy import func, select, update
 from sqlalchemy.orm import Session
 
 from app import quota, sysconfig
-from app.auth import current_user, has_write_scope, require_human
+from app.auth import current_user, require_human, require_write_scope
 from app.db import get_db
+from app.utils import iso
 from app.models import (
     LIVE_OR_UPCOMING,
     BlockedChannel,
@@ -67,7 +68,7 @@ def list_channels(
         ).all():
             agg[cid] = {
                 "stored": n,
-                "last_video_at": last_up.isoformat() if last_up else None,
+                "last_video_at": iso(last_up),
                 "total_duration_seconds": int(total_dur or 0),
                 "count_short": int(shorts),
                 "count_live": int(live),
@@ -232,7 +233,7 @@ def _channel_detail_dict(
         "subscriber_count": channel.subscriber_count,
         "video_count": channel.video_count,
         "total_view_count": channel.total_view_count,
-        "published_at": channel.published_at.isoformat() if channel.published_at else None,
+        "published_at": iso(channel.published_at),
         "external_links": channel.external_links or [],
         "country": channel.country,
         "default_language": channel.default_language,
@@ -454,19 +455,14 @@ def reset_backfill(
 @router.post("/{channel_id}/subscribe")
 def subscribe(
     channel_id: str,
-    user: User = Depends(current_user),
+    user: User = Depends(require_write_scope),
     db: Session = Depends(get_db),
 ) -> dict:
     """Subscribe to a channel discovered from the user's playlists. Gated behind the
-    optional write scope (YouTube subscriptions.insert). Creates the local subscription
-    with YouTube's returned resource id and enriches the channel's metadata if it's still
-    a stub, so the new row renders properly in the manager. It does NOT pull the channel's
-    videos — the scheduler picks those up on its next run, like any subscribed channel."""
-    if not has_write_scope(user):
-        raise HTTPException(
-            status_code=403,
-            detail="Enable playlist editing in Settings to subscribe on YouTube.",
-        )
+    optional write scope (YouTube subscriptions.insert) via require_write_scope. Creates the
+    local subscription with YouTube's returned resource id and enriches the channel's metadata
+    if it's still a stub, so the new row renders properly in the manager. It does NOT pull the
+    channel's videos — the scheduler picks those up on its next run, like any subscribed channel."""
     channel = db.get(Channel, channel_id)
     if channel is None:
         raise HTTPException(status_code=404, detail="Unknown channel")
@@ -524,16 +520,12 @@ def subscribe(
 @router.delete("/{channel_id}/subscription")
 def unsubscribe(
     channel_id: str,
-    user: User = Depends(current_user),
+    user: User = Depends(require_write_scope),
     db: Session = Depends(get_db),
 ) -> dict:
     """Unsubscribe from this channel on YouTube and drop the local subscription. Gated
-    behind the optional write scope; read-only users get a 403 and should Hide instead."""
-    if not has_write_scope(user):
-        raise HTTPException(
-            status_code=403,
-            detail="Enable playlist editing in Settings to unsubscribe on YouTube.",
-        )
+    behind the optional write scope (require_write_scope); read-only users get a 403 and
+    should Hide instead."""
     sub = _user_subscription(db, user, channel_id)
     if not sub.yt_subscription_id:
         raise HTTPException(

backend/app/routes/downloads.py

@@ -18,6 +18,7 @@ from app import sysconfig
 from app.auth import admin_user, require_human
 from app.config import settings
 from app.db import get_db
+from app.utils import iso
 from app.downloads import edit as editmod
 from app.downloads import links as linksmod
 from app.downloads import formats, quota, service, storage
@@ -80,7 +81,7 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
         "phase": job.phase,
         "error": job.error,
         "queue_pos": job.queue_pos,
-        "created_at": job.created_at.isoformat() if job.created_at else None,
+        "created_at": iso(job.created_at),
         "source_kind": job.source_kind,
         "source_ref": job.source_ref,
         "source_url": service.reference_url(job, asset),
@@ -106,7 +107,7 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
             "status": asset.status,
             "title": asset.title,
             "uploader": asset.uploader,
-            "upload_date": asset.upload_date.isoformat() if asset.upload_date else None,
+            "upload_date": iso(asset.upload_date),
             "duration_s": asset.duration_s,
             "size_bytes": asset.size_bytes,
             "width": asset.width,
@@ -114,7 +115,7 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
             "container": asset.container,
             "uploader_url": asset.uploader_url,
             "thumbnail_url": asset.thumbnail_url,
-            "expires_at": asset.expires_at.isoformat() if asset.expires_at else None,
+            "expires_at": iso(asset.expires_at),
         },
     }
 

backend/app/routes/feed.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm.exc import StaleDataError
 from app import quota
 from app.auth import current_user
 from app.db import get_db
+from app.utils import iso
 from app.models import (
     LIVE_OR_UPCOMING,
     BlockedChannel,
@@ -97,7 +98,7 @@ def _serialize(row) -> dict:
         "channel_title": row.channel_title,
         "channel_thumbnail": row.channel_thumbnail,
         "channel_url": _channel_url(row.channel_id, row.channel_handle),
-        "published_at": row.published_at.isoformat() if row.published_at else None,
+        "published_at": iso(row.published_at),
         "thumbnail_url": row.thumbnail_url,
         "duration_seconds": row.duration_seconds,
         "view_count": row.view_count,
@@ -732,7 +733,7 @@ def get_video_detail(
             "in_db": True,
             "channel_id": v.channel_id,
             "channel_title": v.channel.title if v.channel else None,
-            "published_at": v.published_at.isoformat() if v.published_at else None,
+            "published_at": iso(v.published_at),
             "view_count": v.view_count,
             "duration_seconds": v.duration_seconds,
         }

backend/app/routes/messages.py

@@ -27,6 +27,7 @@ from app.ratelimit import RateLimiter
 from app.realtime import manager
 from app.security import verify_password
 from app.userscope import is_messageable_user, messageable_clauses
+from app.utils import iso
 
 router = APIRouter(prefix="/api/messages", tags=["messages"])
 
@@ -97,8 +98,8 @@ def _serialize_msg(m: Message) -> dict:
         "body": m.body,  # only set for system messages
         "ciphertext": m.ciphertext,
         "iv": m.iv,
-        "read_at": m.read_at.isoformat() if m.read_at else None,
-        "created_at": m.created_at.isoformat() if m.created_at else None,
+        "read_at": iso(m.read_at),
+        "created_at": iso(m.created_at),
     }
 
 

backend/app/routes/notifications.py

@@ -11,6 +11,7 @@ from sqlalchemy.orm import Session
 from app.auth import current_user
 from app.db import get_db
 from app.models import Notification, User
+from app.utils import iso
 from app.notifications import trim_read
 
 router = APIRouter(prefix="/api/me/notifications", tags=["notifications"])
@@ -25,7 +26,7 @@ def _serialize(n: Notification) -> dict:
         "data": n.data,
         "read": n.read,
         "dismissed": n.dismissed,
-        "created_at": n.created_at.isoformat() if n.created_at else None,
+        "created_at": iso(n.created_at),
     }
 
 

backend/app/routes/playlists.py

@@ -6,7 +6,7 @@ from sqlalchemy import func, select, and_
 from sqlalchemy.orm import Session, aliased
 
 from app import quota
-from app.auth import current_user, has_read_scope, has_write_scope
+from app.auth import current_user, has_read_scope, has_write_scope, require_write_scope
 from app.db import get_db
 from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState
 from app.routes.feed import _serialize, feed_columns
@@ -316,16 +316,13 @@ def _pushable(db: Session, user: User, playlist_id: int) -> Playlist:
 @router.get("/{playlist_id}/push-plan")
 def push_plan(
     playlist_id: int,
-    user: User = Depends(current_user),
+    user: User = Depends(require_write_scope),
     db: Session = Depends(get_db),
 ) -> dict:
     """Dry run: what 'Sync to YouTube' would do + the quota estimate, so the UI can show a
-    confirm. For a linked playlist this reads the live YouTube state (cheap, 1 unit/page)."""
+    confirm. For a linked playlist this reads the live YouTube state (cheap, 1 unit/page).
+    Write scope is required (require_write_scope) — syncing is a YouTube write."""
     pl = _pushable(db, user, playlist_id)
-    if not has_write_scope(user):
-        raise HTTPException(
-            status_code=403, detail="Enable YouTube editing in Settings first."
-        )
     try:
         with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PUSH):
             plan = plan_push(db, user, pl)
@@ -339,17 +336,13 @@ def push_plan(
 @router.post("/{playlist_id}/push")
 def push(
     playlist_id: int,
-    user: User = Depends(current_user),
+    user: User = Depends(require_write_scope),
     db: Session = Depends(get_db),
 ) -> dict:
     """Push a local playlist to YouTube (create it on first push, else reconcile to match
     local — local wins). Refuses if the estimated quota exceeds what's left today, so a
-    big playlist can't strand half-pushed."""
+    big playlist can't strand half-pushed. Write scope required (require_write_scope)."""
     pl = _pushable(db, user, playlist_id)
-    if not has_write_scope(user):
-        raise HTTPException(
-            status_code=403, detail="Enable YouTube editing in Settings first."
-        )
     try:
         with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PUSH):
             # Read the live playlist ONCE (linked playlists only) and share it with both the

backend/app/routes/plex.py

@@ -41,6 +41,8 @@ from app.models import (
     PlexState,
     User,
 )
+from app.routes._common import owned_or_404
+from app.utils import iso
 from app.plex import paths as plex_paths
 from app.plex import people as plex_people
 from app.plex import stream as plex_stream
@@ -141,11 +143,7 @@ def _link_dict(link: PlexLink | None) -> dict:
         "sync_enabled": bool(link and link.sync_enabled),
         "uses_admin": bool(link and link.uses_admin),
         "initial_import_done": bool(link and link.initial_import_done),
-        "last_watch_sync_at": (
-            link.last_watch_sync_at.isoformat()
-            if link and link.last_watch_sync_at
-            else None
-        ),
+        "last_watch_sync_at": iso(link.last_watch_sync_at) if link else None,
     }
 
 
@@ -912,10 +910,7 @@ def collection_set_editable(
 
 # --- Playlists (Siftlode-native, per-user, ordered — no Plex account needed; Plex sync is a later phase).
 def _own_playlist_or_404(db: Session, pid: int, user: User) -> PlexPlaylist:
-    pl = db.query(PlexPlaylist).filter_by(id=pid, user_id=user.id).first()
-    if pl is None:
-        raise HTTPException(status_code=404, detail="Playlist not found")
-    return pl
+    return owned_or_404(db, PlexPlaylist, pid, user, detail="Playlist not found")
 
 
 def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict:

backend/app/routes/saved_views.py

@@ -6,6 +6,7 @@ from sqlalchemy.orm import Session
 from app.auth import require_human
 from app.db import get_db
 from app.models import SavedView, User
+from app.routes._common import owned_or_404
 
 # Saved "smart views": per-user named snapshots of the feed's FeedFilters. Gated behind
 # require_human so the shared demo account can't pollute a communal list (mirrors how other
@@ -71,10 +72,7 @@ def create_view(
 
 
 def _own_view(db: Session, user: User, view_id: int) -> SavedView:
-    view = db.get(SavedView, view_id)
-    if view is None or view.user_id != user.id:
-        raise HTTPException(status_code=404, detail="Unknown view")
-    return view
+    return owned_or_404(db, SavedView, view_id, user, detail="Unknown view")
 
 
 @router.patch("/{view_id}")

backend/app/utils.py

@@ -2,6 +2,13 @@
 import re
 from datetime import datetime, timezone
 
+
+def iso(dt: datetime | None) -> str | None:
+    """Serialize a datetime for a JSON response: ISO-8601, or None when the column is NULL.
+    Single source of truth for the `x.isoformat() if x else None` idiom the routes had
+    hand-rolled ~17 times — so a future tz/format tweak lands in one place."""
+    return dt.isoformat() if dt is not None else None
+
 # Single source of truth for "looks like an email" — used by every endpoint that accepts an
 # address (auth register/reset/demo, admin invites + demo whitelist, the setup wizard). Keep
 # the strictness in one place so routes don't drift between this and a looser "@"-in check.

backend/tests/test_utils.py

@@ -0,0 +1,25 @@
+"""iso(): the single datetime→JSON serializer the routes share (R6 S1a)."""
+from datetime import datetime, timezone
+
+from app.utils import iso
+
+
+def test_none_maps_to_none():
+    assert iso(None) is None
+
+
+def test_aware_datetime_keeps_offset():
+    dt = datetime(2026, 7, 26, 12, 30, 0, tzinfo=timezone.utc)
+    assert iso(dt) == "2026-07-26T12:30:00+00:00"
+
+
+def test_naive_datetime_serializes_without_offset():
+    dt = datetime(2026, 7, 26, 12, 30, 0)
+    assert iso(dt) == "2026-07-26T12:30:00"
+
+
+def test_matches_datetime_isoformat_contract():
+    # iso is exactly `dt.isoformat()` for a present value — guards the routes that used to
+    # hand-roll `x.isoformat() if x else None`.
+    dt = datetime(2020, 1, 2, 3, 4, 5, 678901, tzinfo=timezone.utc)
+    assert iso(dt) == dt.isoformat()