refactor(api): Pydantic request models for downloads routes (R6 S1b)

All 11 download endpoints off payload:dict. Highlights:
- admin_set_quota: AdminSetQuotaIn — the `int(payload["max_bytes"])` 500 on a
  non-number is now a 422; each field set only when provided (is not None).
- profiles/enqueue/edit: CreateProfileIn / UpdateProfileIn / EnqueueDownloadIn /
  EnqueueEditIn; spec/edit_spec typed as dict (non-object → 422). _resolve_spec now
  takes (profile_id, spec) instead of the raw payload.
- meta/share/links: UpdateDownloadMetaIn / ShareDownloadIn / CreateLinkIn /
  UpdateLinkIn; _link_expiry takes an int|None instead of digging the payload.
PATCH endpoints read model_fields_set so an explicit null still clears a field.

Gate: ruff clean, pytest 164 green.

backend/app/routes/downloads.py

@@ -11,6 +11,7 @@ from urllib.parse import parse_qs, urlparse
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import FileResponse
+from pydantic import BaseModel
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
@@ -173,16 +174,19 @@ def list_profiles(
     return [_serialize_profile(p) for p in rows]
 
 
+class CreateProfileIn(BaseModel):
+    name: str
+    spec: dict
+
+
 @router.post("/profiles")
 def create_profile(
-    payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
+    body: CreateProfileIn, user: User = Depends(require_human), db: Session = Depends(get_db)
 ) -> dict:
-    name = (payload.get("name") or "").strip()
+    name = body.name.strip()
-    spec = payload.get("spec")
+    spec = body.spec
     if not name:
         raise HTTPException(status_code=400, detail="A profile name is required.")
-    if not isinstance(spec, dict):
-        raise HTTPException(status_code=400, detail="spec must be an object")
     maxpos = db.execute(
         select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where(
             DownloadProfile.user_id == user.id
@@ -196,25 +200,31 @@ def create_profile(
     return _serialize_profile(p)
 
 
+class UpdateProfileIn(BaseModel):
+    name: str | None = None
+    spec: dict | None = None
+
+
 @router.patch("/profiles/{profile_id}")
 def update_profile(
     profile_id: int,
-    payload: dict,
+    body: UpdateProfileIn,
     user: User = Depends(require_human),
     db: Session = Depends(get_db),
 ) -> dict:
     p = db.get(DownloadProfile, profile_id)
     if p is None or p.is_builtin or p.user_id != user.id:
         raise HTTPException(status_code=404, detail="Unknown profile")
-    if "name" in payload:
+    sent = body.model_fields_set
-        name = (payload.get("name") or "").strip()
+    if "name" in sent:
+        name = (body.name or "").strip()
         if not name:
             raise HTTPException(status_code=400, detail="A profile name is required.")
         p.name = name[:80]
-    if "spec" in payload:
+    if "spec" in sent:
-        if not isinstance(payload["spec"], dict):
+        if body.spec is None:
             raise HTTPException(status_code=400, detail="spec must be an object")
-        p.spec = payload["spec"]
+        p.spec = body.spec
     db.commit()
     return _serialize_profile(p)
 
@@ -231,17 +241,18 @@ def delete_profile(
     return {"deleted": profile_id}
 
 
-def _resolve_spec(db: Session, user: User, payload: dict) -> tuple[dict, int | None]:
+def _resolve_spec(
+    db: Session, user: User, profile_id: int | None, spec: dict | None
+) -> tuple[dict, int | None]:
     """Pick the format spec for an enqueue: an explicit profile_id (builtin or the user's own),
     an inline spec, or the first builtin as a fallback."""
-    pid = payload.get("profile_id")
+    if profile_id is not None:
-    if pid is not None:
+        p = db.get(DownloadProfile, profile_id)
-        p = db.get(DownloadProfile, pid)
         if p is None or (not p.is_builtin and p.user_id != user.id):
             raise HTTPException(status_code=404, detail="Unknown profile")
         return p.spec, p.id
-    if isinstance(payload.get("spec"), dict):
+    if spec is not None:
-        return payload["spec"], None
+        return spec, None
     fallback = db.execute(
         select(DownloadProfile)
         .where(DownloadProfile.is_builtin.is_(True))
@@ -255,13 +266,20 @@ def _resolve_spec(db: Session, user: User, payload: dict) -> tuple[dict, int | N
 
 # --- enqueue + list + manage ---------------------------------------------------------------
 
+class EnqueueDownloadIn(BaseModel):
+    source: str = ""
+    profile_id: int | None = None
+    spec: dict | None = None
+    display_name: str | None = None
+
+
 @router.post("")
 def enqueue_download(
-    payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
+    body: EnqueueDownloadIn, user: User = Depends(require_human), db: Session = Depends(get_db)
 ) -> dict:
-    source_kind, source_ref = resolve_source(payload.get("source") or "")
+    source_kind, source_ref = resolve_source(body.source)
-    spec, profile_id = _resolve_spec(db, user, payload)
+    spec, profile_id = _resolve_spec(db, user, body.profile_id, body.spec)
-    display_name = (payload.get("display_name") or "").strip() or None
+    display_name = (body.display_name or "").strip() or None
     try:
         job = service.enqueue(
             db, user.id, source_kind, source_ref, spec, profile_id, display_name
@@ -271,22 +289,26 @@ def enqueue_download(
     return _serialize_job(db, job)
 
 
+class EnqueueEditIn(BaseModel):
+    source_job_id: int = 0
+    edit_spec: dict
+    display_name: str | None = None
+
+
 @router.post("/edit")
 def enqueue_edit(
-    payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
+    body: EnqueueEditIn, user: User = Depends(require_human), db: Session = Depends(get_db)
 ) -> dict:
     """Queue a phase-2 editor derivative (trim/crop) of a download this user can access.
 
     The source may be the user's own download OR one shared with them — either way the resulting
     clip is a NEW per-user derivative in the editor's own library (counts against their quota); the
     source file is only read, never modified, so editing a shared video is safe."""
-    source_job = _accessible_job(db, user, int(payload.get("source_job_id") or 0))
+    source_job = _accessible_job(db, user, body.source_job_id)
     if source_job.status != "done" or source_job.asset_id is None:
         raise HTTPException(status_code=409, detail="You can only edit a finished download.")
-    spec = payload.get("edit_spec")
+    spec = body.edit_spec
-    if not isinstance(spec, dict):
+    display_name = (body.display_name or "").strip() or None
-        raise HTTPException(status_code=400, detail="edit_spec must be an object")
-    display_name = (payload.get("display_name") or "").strip() or None
     try:
         job = service.enqueue_edit(db, user.id, source_job, spec, display_name)
     except quota.QuotaExceeded as e:
@@ -389,26 +411,34 @@ def _clean_url(raw) -> str | None:
     return url[:2048]
 
 
+class UpdateDownloadMetaIn(BaseModel):
+    # PATCH display-only overrides; presence (model_fields_set) decides what's touched. A field
+    # set to null/"" clears the override (falls back to the on-disk system value).
+    display_name: str | None = None
+    display_uploader: str | None = None
+    display_uploader_url: str | None = None
+    extra_links: list | None = None
+
+
 @router.patch("/{job_id}")
 def update_download_meta(
     job_id: int,
-    payload: dict,
+    body: UpdateDownloadMetaIn,
     user: User = Depends(require_human),
     db: Session = Depends(get_db),
 ) -> dict:
     """Update a download's display metadata: title, channel name + link, and extra reference URLs.
     All are display-only overrides — the on-disk file keeps its system-decided, id-bound name."""
     job = _own_job(db, user, job_id)
-    if "display_name" in payload:
+    sent = body.model_fields_set
-        job.display_name = ((payload.get("display_name") or "").strip())[:255] or None
+    if "display_name" in sent:
-    if "display_uploader" in payload:
+        job.display_name = ((body.display_name or "").strip())[:255] or None
-        job.display_uploader = ((payload.get("display_uploader") or "").strip())[:255] or None
+    if "display_uploader" in sent:
-    if "display_uploader_url" in payload:
+        job.display_uploader = ((body.display_uploader or "").strip())[:255] or None
-        job.display_uploader_url = _clean_url(payload.get("display_uploader_url"))
+    if "display_uploader_url" in sent:
-    if "extra_links" in payload:
+        job.display_uploader_url = _clean_url(body.display_uploader_url)
-        raw = payload.get("extra_links") or []
+    if "extra_links" in sent:
-        if not isinstance(raw, list):
+        raw = body.extra_links or []
-            raise HTTPException(status_code=400, detail="extra_links must be a list")
         cleaned = [u for u in (_clean_url(x) for x in raw) if u][:_MAX_EXTRA_LINKS]
         job.extra_links = cleaned or None
     db.commit()
@@ -652,17 +682,21 @@ def storyboard_image(
 
 # --- sharing -------------------------------------------------------------------------------
 
+class ShareDownloadIn(BaseModel):
+    email: str = ""
+
+
 @router.post("/{job_id}/share")
 def share_download(
     job_id: int,
-    payload: dict,
+    body: ShareDownloadIn,
     user: User = Depends(require_human),
     db: Session = Depends(get_db),
 ) -> dict:
     job = _own_job(db, user, job_id)
     if job.status != "done" or job.asset_id is None:
         raise HTTPException(status_code=409, detail="Only a finished download can be shared.")
-    target = (payload.get("email") or "").strip().lower()
+    target = body.email.strip().lower()
     if not target:
         raise HTTPException(status_code=400, detail="A recipient email is required.")
     recipient = db.execute(
@@ -771,17 +805,12 @@ def _own_link(db: Session, user: User, link_id: int) -> DownloadLink:
     return link
 
 
-def _link_expiry(payload: dict) -> datetime | None:
+def _link_expiry(days: int | None) -> datetime | None:
-    days = payload.get("expires_days")
+    # None/0/negative → no expiry; otherwise N days out, capped at ~10 years. The Pydantic model
-    if days in (None, "", 0, "0"):
+    # already coerced/validated the type, so a non-number is a 422 before we get here.
+    if not days or days <= 0:
         return None
-    try:
+    return datetime.now(timezone.utc) + timedelta(days=min(days, 3650))
-        d = int(days)
-    except (TypeError, ValueError):
-        raise HTTPException(status_code=400, detail="expires_days must be a number")
-    if d <= 0:
-        return None
-    return datetime.now(timezone.utc) + timedelta(days=min(d, 3650))
 
 
 @router.get("/{job_id}/links")
@@ -799,24 +828,30 @@ def list_links(
     return [linksmod.owner_view(row) for row in rows]
 
 
+class CreateLinkIn(BaseModel):
+    password: str = ""
+    allow_download: bool = False
+    expires_days: int | None = None
+
+
 @router.post("/{job_id}/links")
 def create_link(
     job_id: int,
-    payload: dict,
+    body: CreateLinkIn,
     user: User = Depends(require_human),
     db: Session = Depends(get_db),
 ) -> dict:
     job = _own_job(db, user, job_id)
     if job.status != "done" or job.asset_id is None:
         raise HTTPException(status_code=409, detail="Only a finished download can be shared.")
-    pw = (payload.get("password") or "").strip()
+    pw = body.password.strip()
     link = DownloadLink(
         job_id=job_id,
         token=linksmod.new_token(),
         created_by=user.id,
-        allow_download=bool(payload.get("allow_download")),
+        allow_download=body.allow_download,
         password_hash=hash_password(pw) if pw else None,
-        expires_at=_link_expiry(payload),
+        expires_at=_link_expiry(body.expires_days),
     )
     db.add(link)
     db.commit()
@@ -824,20 +859,27 @@ def create_link(
     return linksmod.owner_view(link)
 
 
+class UpdateLinkIn(BaseModel):
+    allow_download: bool | None = None
+    expires_days: int | None = None
+    password: str | None = None
+
+
 @router.patch("/links/{link_id}")
 def update_link(
     link_id: int,
-    payload: dict,
+    body: UpdateLinkIn,
     user: User = Depends(require_human),
     db: Session = Depends(get_db),
 ) -> dict:
     link = _own_link(db, user, link_id)
-    if "allow_download" in payload:
+    sent = body.model_fields_set
-        link.allow_download = bool(payload["allow_download"])
+    if "allow_download" in sent:
-    if "expires_days" in payload:
+        link.allow_download = bool(body.allow_download)
-        link.expires_at = _link_expiry(payload)
+    if "expires_days" in sent:
-    if "password" in payload:
+        link.expires_at = _link_expiry(body.expires_days)
-        pw = (payload.get("password") or "").strip()
+    if "password" in sent:
+        pw = (body.password or "").strip()
         link.password_hash = hash_password(pw) if pw else None
         # Bump the version so any grant issued under the previous password stops working now, rather
         # than living out its 6h TTL (setting/clearing/changing the password all count as a rotation).
@@ -933,10 +975,19 @@ def admin_get_quota(
     }
 
 
+class AdminSetQuotaIn(BaseModel):
+    # Each field is set only when provided as a real value (is not None). A non-numeric limit
+    # used to reach int() and 500; it's now a clean 422.
+    max_bytes: int | None = None
+    max_concurrent: int | None = None
+    max_jobs: int | None = None
+    unlimited: bool | None = None
+
+
 @admin_router.put("/quota/{user_id}")
 def admin_set_quota(
     user_id: int,
-    payload: dict,
+    body: AdminSetQuotaIn,
     admin: User = Depends(admin_user),
     db: Session = Depends(get_db),
 ) -> dict:
@@ -953,14 +1004,14 @@ def admin_set_quota(
             unlimited=cur.unlimited,
         )
         db.add(row)
-    if "max_bytes" in payload:
+    if body.max_bytes is not None:
-        row.max_bytes = max(0, int(payload["max_bytes"]))
+        row.max_bytes = max(0, body.max_bytes)
-    if "max_concurrent" in payload:
+    if body.max_concurrent is not None:
-        row.max_concurrent = max(1, int(payload["max_concurrent"]))
+        row.max_concurrent = max(1, body.max_concurrent)
-    if "max_jobs" in payload:
+    if body.max_jobs is not None:
-        row.max_jobs = max(1, int(payload["max_jobs"]))
+        row.max_jobs = max(1, body.max_jobs)
-    if "unlimited" in payload:
+    if body.unlimited is not None:
-        row.unlimited = bool(payload["unlimited"])
+        row.unlimited = body.unlimited
     db.commit()
     return admin_get_quota(user_id, admin, db)