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

CreatePlaylistIn / RenamePlaylistIn (PATCH via model_fields_set) / ReorderItemsIn,
plus a shared VideoRefIn(video_id) for both add-video endpoints (watch-later +
add-item). A non-string video_id is now a 422 instead of reaching db.get.

Gate: ruff clean.
This commit is contained in:
2026-07-26 04:58:37 +02:00
parent ab83aae8bb
commit 8bf5026e7c
+31 -13
View File
@@ -2,6 +2,7 @@
catalog. Foundation phase = local only (create / rename / delete, add / remove / reorder catalog. Foundation phase = local only (create / rename / delete, add / remove / reorder
items). YouTube two-way sync (mirror existing playlists, push edits) lands in later phases.""" items). YouTube two-way sync (mirror existing playlists, push edits) lands in later phases."""
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import func, select, and_ from sqlalchemy import func, select, and_
from sqlalchemy.orm import Session, aliased from sqlalchemy.orm import Session, aliased
@@ -197,13 +198,22 @@ def list_playlists(
] ]
class CreatePlaylistIn(BaseModel):
name: str
# Shared by the two "add a video" endpoints (Watch-later shortcut + explicit add-to-playlist).
class VideoRefIn(BaseModel):
video_id: str
@router.post("") @router.post("")
def create_playlist( def create_playlist(
payload: dict, body: CreatePlaylistIn,
user: User = Depends(current_user), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> dict:
name = (payload.get("name") or "").strip() name = body.name.strip()
if not name: if not name:
raise HTTPException(status_code=400, detail="name is required") raise HTTPException(status_code=400, detail="name is required")
pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id)) pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id))
@@ -233,13 +243,13 @@ def _watch_later(db: Session, user: User) -> Playlist:
@router.post("/watch-later") @router.post("/watch-later")
def add_watch_later( def add_watch_later(
payload: dict, body: VideoRefIn,
user: User = Depends(current_user), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> dict:
"""Bookmark shortcut: add a video to the built-in Watch later playlist.""" """Bookmark shortcut: add a video to the built-in Watch later playlist."""
video_id = payload.get("video_id") video_id = body.video_id
if not video_id or db.get(Video, video_id) is None: if db.get(Video, video_id) is None:
raise HTTPException(status_code=404, detail="Unknown video") raise HTTPException(status_code=404, detail="Unknown video")
pl = _watch_later(db, user) pl = _watch_later(db, user)
_add_item(db, pl, video_id, mark_dirty=False) _add_item(db, pl, video_id, mark_dirty=False)
@@ -394,16 +404,20 @@ def get_playlist(
} }
class RenamePlaylistIn(BaseModel):
name: str | None = None
@router.patch("/{playlist_id}") @router.patch("/{playlist_id}")
def rename_playlist( def rename_playlist(
playlist_id: int, playlist_id: int,
payload: dict, body: RenamePlaylistIn,
user: User = Depends(current_user), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> dict:
pl = _own_playlist(db, user, playlist_id) pl = _own_playlist(db, user, playlist_id)
if "name" in payload: if "name" in body.model_fields_set:
name = (payload.get("name") or "").strip() name = (body.name or "").strip()
if not name: if not name:
raise HTTPException(status_code=400, detail="name cannot be empty") raise HTTPException(status_code=400, detail="name cannot be empty")
if name != pl.name: if name != pl.name:
@@ -462,13 +476,13 @@ def delete_playlist(
@router.post("/{playlist_id}/items") @router.post("/{playlist_id}/items")
def add_item( def add_item(
playlist_id: int, playlist_id: int,
payload: dict, body: VideoRefIn,
user: User = Depends(current_user), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> dict:
pl = _own_playlist(db, user, playlist_id) pl = _own_playlist(db, user, playlist_id)
video_id = payload.get("video_id") video_id = body.video_id
if not video_id or db.get(Video, video_id) is None: if db.get(Video, video_id) is None:
raise HTTPException(status_code=404, detail="Unknown video") raise HTTPException(status_code=404, detail="Unknown video")
_add_item(db, pl, video_id, mark_dirty=True) _add_item(db, pl, video_id, mark_dirty=True)
return {"playlist_id": pl.id, "video_id": video_id} return {"playlist_id": pl.id, "video_id": video_id}
@@ -486,15 +500,19 @@ def remove_item(
return {"playlist_id": pl.id, "video_id": video_id} return {"playlist_id": pl.id, "video_id": video_id}
class ReorderItemsIn(BaseModel):
video_ids: list[str] = []
@router.put("/{playlist_id}/order") @router.put("/{playlist_id}/order")
def reorder_items( def reorder_items(
playlist_id: int, playlist_id: int,
payload: dict, body: ReorderItemsIn,
user: User = Depends(current_user), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> dict:
pl = _own_playlist(db, user, playlist_id) pl = _own_playlist(db, user, playlist_id)
order = payload.get("video_ids") or [] order = body.video_ids
items = { items = {
it.video_id: it it.video_id: it
for it in db.execute( for it in db.execute(