feat(playlists): YouTube write API client methods

Add OAuth-only write helpers to YouTubeClient (each 50 quota units):
create_playlist, add_playlist_item, move_playlist_item, delete_playlist_item,
delete_playlist, plus iter_playlist_items_with_ids for diffing. A shared _write
helper records quota and raises YouTubeError on non-2xx.
This commit is contained in:
npeter83 2026-06-15 21:22:51 +02:00
parent 3803ae00fe
commit 2b79ba6637

View file

@ -213,6 +213,93 @@ class YouTubeClient:
f"DELETE subscriptions -> {resp.status_code}: {resp.text[:300]}"
)
def iter_playlist_items_with_ids(self, playlist_id: str) -> Iterator[dict]:
"""Yield {item_id, video_id} for each item of one of the user's playlists, in
playlist order. `item_id` is the playlistItem resource id, needed to delete or
reposition the item via the write API. OAuth (private playlists work)."""
page_token = None
while True:
params = {
"part": "contentDetails",
"playlistId": playlist_id,
"maxResults": 50,
}
if page_token:
params["pageToken"] = page_token
data = self._get("playlistItems", params, cost=1, allow_key=False)
for item in data.get("items", []):
vid = item.get("contentDetails", {}).get("videoId")
if vid:
yield {"item_id": item.get("id"), "video_id": vid}
page_token = data.get("nextPageToken")
if not page_token:
break
# --- write endpoints (OAuth only, never the API key; each costs 50 quota units) ---
def _write(self, method: str, path: str, *, params: dict, json: dict | None = None) -> dict:
headers = {"Authorization": f"Bearer {self._access_token()}"}
resp = self._http.request(
method, f"{API_BASE}/{path}", params=params, json=json, headers=headers
)
quota.record_usage(self.db, 50)
if resp.status_code not in (200, 204):
log.warning("YouTube %s %s -> %s: %s", method, path, resp.status_code, resp.text[:200])
raise YouTubeError(f"{method} {path} -> {resp.status_code}: {resp.text[:300]}")
return resp.json() if resp.content else {}
def create_playlist(self, title: str, description: str = "", privacy: str = "private") -> str:
"""Create a new YouTube playlist; returns its id. 50 units."""
data = self._write(
"POST",
"playlists",
params={"part": "snippet,status"},
json={
"snippet": {"title": title[:150], "description": description[:5000]},
"status": {"privacyStatus": privacy},
},
)
return data["id"]
def add_playlist_item(self, playlist_id: str, video_id: str, position: int | None = None) -> str:
"""Append (or insert at `position`) a video into a playlist; returns the new
playlistItem id. 50 units."""
snippet = {
"playlistId": playlist_id,
"resourceId": {"kind": "youtube#video", "videoId": video_id},
}
if position is not None:
snippet["position"] = position
data = self._write(
"POST", "playlistItems", params={"part": "snippet"}, json={"snippet": snippet}
)
return data["id"]
def move_playlist_item(
self, item_id: str, playlist_id: str, video_id: str, position: int
) -> None:
"""Reposition an existing playlistItem to absolute 0-based `position`. 50 units."""
self._write(
"PUT",
"playlistItems",
params={"part": "snippet"},
json={
"id": item_id,
"snippet": {
"playlistId": playlist_id,
"position": position,
"resourceId": {"kind": "youtube#video", "videoId": video_id},
},
},
)
def delete_playlist_item(self, item_id: str) -> None:
"""Remove a playlistItem from its playlist. 50 units."""
self._write("DELETE", "playlistItems", params={"id": item_id})
def delete_playlist(self, playlist_id: str) -> None:
"""Delete a whole playlist on YouTube. 50 units."""
self._write("DELETE", "playlists", params={"id": playlist_id})
def get_videos(self, video_ids: list[str]) -> list[dict]:
items: list[dict] = []
for batch in _chunks(video_ids, 50):