fix: remove a deleted video's leftover entry from a playlist

src/youtube_kit/core/ops/playlists.py

@@ -33,6 +33,7 @@
 from .channels import copied_from_recent_uploads, my_channel_id, owner_or
 
 _PARTS = "snippet,status,contentDetails"
+_MAX_PLAYLIST_LENGTH = 5000  # YouTube's cap on a user playlist (uploads and Watch Later can be longer)
 
 
 @operation(quota=1, read_only=True, idempotent=True, paginated=True)
@@ -200,14 +201,18 @@ def add_video_to_playlist(playlist: str, video: str, *, position: int | None = N
 def remove_from_playlist(playlist: str, video: str) -> PlaylistRemoval:
     """Remove a video from one of your playlists, every time it appears. The video itself is not affected.
 
+    YouTube can't look up entries of a deleted video, so for a deleted or unknown video the whole playlist is read:
+    1 more unit per 50 entries.
+
     Args:
         playlist: Playlist ID or URL.
         video: Video ID or URL.
     """
     playlist_id, video_id = parse.playlist_id(playlist), parse.video_id(video)
     youtube = client.oauth()
     removed = sum(
-        delete_if_present(youtube.playlistItems().delete(id=item["id"])) for item in _occurrences(playlist_id, video_id)
+        delete_if_present(youtube.playlistItems().delete(id=item["id"]))
+        for item in _occurrences(playlist_id, video_id, scan_if_deleted=True)
     )
     return PlaylistRemoval(playlist_id=playlist_id, video_id=video_id, removed=removed)
 
@@ -280,14 +285,31 @@ def set_playlist_thumbnail(playlist: str, image: Path) -> PlaylistImage:
     )
 
 
-def _occurrences(playlist_id: str, video_id: str) -> list[dict]:
+def _occurrences(playlist_id: str, video_id: str, *, scan_if_deleted: bool = False) -> list[dict]:
     """The playlist items holding a video; more than one if it was added repeatedly."""
-    request = (
-        client.oauth()
-        .playlistItems()
-        .list(part="id", playlistId=playlist_id, videoId=video_id, maxResults=API_PAGE_SIZE)
+    items = client.oauth().playlistItems()
+    try:
+        return (
+            items.list(part="id", playlistId=playlist_id, videoId=video_id, maxResults=API_PAGE_SIZE)
+            .execute()
+            .get("items", [])
+        )
+    except HttpError as e:
+        if not scan_if_deleted or http_reason(e) != "videoNotFound":
+            raise
+    # The videoId filter fails once a video is deleted, though the playlist keeps its entry: read the whole playlist.
+    page = collect(
+        lambda **page: items.list(part="contentDetails", playlistId=playlist_id, **page),
+        limit=_MAX_PLAYLIST_LENGTH,
+        page_token=None,
+        to_item=lambda item: item,
     )
-    return request.execute().get("items", [])
+    if page.next_page_token:
+        raise YouTubeError(
+            f"Playlist {playlist_id} has more than {_MAX_PLAYLIST_LENGTH} entries, "
+            f"too many to search for video {video_id}. Remove it in YouTube Studio."
+        )
+    return [item for item in page.items if item.get("contentDetails", {}).get("videoId") == video_id]
 
 
 def _item_snippet(playlist_id: str, video_id: str, position: int | None) -> dict:

tests/ops/test_playlists.py

@@ -135,6 +135,71 @@ def test_remove_from_playlist_is_idempotent(api):
     assert playlists.remove_from_playlist("PL1", "dQw4w9WgXcQ").removed == 0
 
 
+def _entries(*video_ids, token=None):
+    items = [{"id": f"item-{v}", "contentDetails": {"videoId": v}} for v in video_ids]
+    return {"items": items, "nextPageToken": token} if token else {"items": items}
+
+
+VIDEO_NOT_FOUND = http_error(404, "videoNotFound", "Video not found.")
+
+
+def test_remove_from_playlist_finds_a_deleted_video_by_reading_the_playlist(api):
+    # YouTube rejects the videoId filter once the video is deleted, though the playlist still holds it.
+    first_page = _entries("gKcKEGRfoAM", "7A22uuMVA6w", token="next")
+    first_page["items"].append({"id": "item-no-video", "contentDetails": {}})
+    items = api.oauth.playlistItems()
+    items.list().execute.side_effect = [VIDEO_NOT_FOUND, first_page, _entries("7A22uuMVA6w")]
+    items.list.reset_mock()
+
+    result = playlists.remove_from_playlist("PL1", "7A22uuMVA6w")
+
+    assert result.removed == 2
+    deleted = [call.kwargs["id"] for call in items.delete.call_args_list if call.kwargs]
+    assert deleted == ["item-7A22uuMVA6w", "item-7A22uuMVA6w"]
+    filtered, *scans = [call.kwargs for call in items.list.call_args_list]
+    assert filtered["videoId"] == "7A22uuMVA6w"
+    assert scans == [
+        {"part": "contentDetails", "playlistId": "PL1", "maxResults": 50},
+        {"part": "contentDetails", "playlistId": "PL1", "maxResults": 50, "pageToken": "next"},
+    ]
+
+
+def test_remove_from_playlist_refuses_a_scan_cut_off_at_the_playlist_cap(api, monkeypatch):
+    monkeypatch.setattr(playlists, "_MAX_PLAYLIST_LENGTH", 1)
+    api.oauth.playlistItems().list().execute.side_effect = [VIDEO_NOT_FOUND, _entries("gKcKEGRfoAM", token="next")]
+
+    with pytest.raises(YouTubeError, match=r"more than 1 entries.*YouTube Studio"):
+        playlists.remove_from_playlist("PL1", "7A22uuMVA6w")
+    api.oauth.playlistItems().delete.assert_not_called()
+
+
+def test_remove_from_playlist_of_a_video_in_no_entry_removes_nothing(api):
+    api.oauth.playlistItems().list().execute.side_effect = [VIDEO_NOT_FOUND, _entries("gKcKEGRfoAM")]
+
+    assert playlists.remove_from_playlist("PL1", "7A22uuMVA6w").removed == 0
+    api.oauth.playlistItems().delete.assert_not_called()
+
+
+def test_remove_from_playlist_raises_other_errors_without_reading_the_playlist(api):
+    items = api.oauth.playlistItems()
+    items.list().execute.side_effect = http_error(404, "playlistNotFound", "Playlist not found.")
+    items.list.reset_mock()
+
+    with pytest.raises(NotFound):
+        playlists.remove_from_playlist("PL1", "7A22uuMVA6w")
+    assert items.list.call_count == 1
+
+
+def test_move_in_playlist_does_not_read_the_playlist_for_a_deleted_video(api):
+    items = api.oauth.playlistItems()
+    items.list().execute.side_effect = VIDEO_NOT_FOUND
+    items.list.reset_mock()
+
+    with pytest.raises(NotFound):
+        playlists.move_in_playlist("PL1", "7A22uuMVA6w", position=0)
+    assert items.list.call_count == 1
+
+
 def test_move_in_playlist(api):
     api.oauth.playlistItems().list().execute.return_value = {"items": [{"id": "item1"}]}
     api.oauth.playlistItems().update().execute.return_value = playlist_item_resource(position=0)

tests/snapshots/mcp/remove_from_playlist.json

@@ -6,7 +6,7 @@
     "read_only_hint": false,
     "title": null
   },
-  "description": "Remove a video from one of your playlists, every time it appears. The video itself is not affected.\n\nQuota: 51 units.",
+  "description": "Remove a video from one of your playlists, every time it appears. The video itself is not affected.\n\nYouTube can't look up entries of a deleted video, so for a deleted or unknown video the whole playlist is read:\n1 more unit per 50 entries.\n\nQuota: 51 units.",
   "input_schema": {
     "additionalProperties": false,
     "properties": {