Rockbox open source high quality audio player as a Music Player Daemon
mpris
rockbox
mpd
libadwaita
audio
rust
zig
deno
1"""Persistent ("saved") playlists and their folders."""
2
3from __future__ import annotations
4
5from dataclasses import asdict, dataclass
6
7from ..transport import HttpTransport
8from ..types import SavedPlaylist, SavedPlaylistFolder
9
10_PLAYLIST_FIELDS = "id name description image folderId trackCount createdAt updatedAt"
11_FOLDER_FIELDS = "id name createdAt updatedAt"
12
13
14@dataclass
15class CreatePlaylistInput:
16 name: str
17 description: str | None = None
18 image: str | None = None
19 folder_id: str | None = None
20 track_ids: list[str] | None = None
21
22
23@dataclass
24class UpdatePlaylistInput:
25 name: str
26 description: str | None = None
27 image: str | None = None
28 folder_id: str | None = None
29
30
31def _camelize(d: dict[str, object]) -> dict[str, object]:
32 """folder_id → folderId, track_ids → trackIds, etc."""
33 out: dict[str, object] = {}
34 for k, v in d.items():
35 if v is None:
36 continue
37 parts = k.split("_")
38 camel = parts[0] + "".join(p.title() for p in parts[1:])
39 out[camel] = v
40 return out
41
42
43class SavedPlaylistsApi:
44 def __init__(self, http: HttpTransport) -> None:
45 self._http = http
46
47 async def list(self, folder_id: str | None = None) -> list[SavedPlaylist]:
48 data = await self._http.execute(
49 "query SavedPlaylists($folderId: String) "
50 f"{{ savedPlaylists(folderId: $folderId) {{ {_PLAYLIST_FIELDS} }} }}",
51 {"folderId": folder_id},
52 )
53 return [SavedPlaylist.model_validate(p) for p in data.get("savedPlaylists", [])]
54
55 async def get(self, id: str) -> SavedPlaylist | None:
56 data = await self._http.execute(
57 "query SavedPlaylist($id: String!) "
58 f"{{ savedPlaylist(id: $id) {{ {_PLAYLIST_FIELDS} }} }}",
59 {"id": id},
60 )
61 raw = data.get("savedPlaylist")
62 return SavedPlaylist.model_validate(raw) if raw is not None else None
63
64 async def track_ids(self, playlist_id: str) -> list[str]:
65 data = await self._http.execute(
66 "query SavedPlaylistTrackIds($playlistId: String!) "
67 "{ savedPlaylistTrackIds(playlistId: $playlistId) }",
68 {"playlistId": playlist_id},
69 )
70 return list(data.get("savedPlaylistTrackIds", []))
71
72 async def create(self, input: CreatePlaylistInput) -> SavedPlaylist:
73 data = await self._http.execute(
74 "mutation CreateSavedPlaylist("
75 "$name: String!, $description: String, $image: String, "
76 "$folderId: String, $trackIds: [String!]) "
77 "{ createSavedPlaylist("
78 "name: $name, description: $description, image: $image, "
79 "folderId: $folderId, trackIds: $trackIds) "
80 f"{{ {_PLAYLIST_FIELDS} }} }}",
81 _camelize(asdict(input)),
82 )
83 return SavedPlaylist.model_validate(data["createSavedPlaylist"])
84
85 async def update(self, id: str, input: UpdatePlaylistInput) -> None:
86 await self._http.execute(
87 "mutation UpdateSavedPlaylist("
88 "$id: String!, $name: String!, $description: String, "
89 "$image: String, $folderId: String) "
90 "{ updateSavedPlaylist("
91 "id: $id, name: $name, description: $description, "
92 "image: $image, folderId: $folderId) }",
93 {"id": id, **_camelize(asdict(input))},
94 )
95
96 async def delete(self, id: str) -> None:
97 await self._http.execute(
98 "mutation DeleteSavedPlaylist($id: String!) { deleteSavedPlaylist(id: $id) }",
99 {"id": id},
100 )
101
102 async def add_tracks(self, playlist_id: str, track_ids: list[str]) -> None:
103 await self._http.execute(
104 "mutation AddTracksToSavedPlaylist($playlistId: String!, $trackIds: [String!]!) "
105 "{ addTracksToSavedPlaylist(playlistId: $playlistId, trackIds: $trackIds) }",
106 {"playlistId": playlist_id, "trackIds": track_ids},
107 )
108
109 async def remove_track(self, playlist_id: str, track_id: str) -> None:
110 await self._http.execute(
111 "mutation RemoveTrackFromSavedPlaylist($playlistId: String!, $trackId: String!) "
112 "{ removeTrackFromSavedPlaylist(playlistId: $playlistId, trackId: $trackId) }",
113 {"playlistId": playlist_id, "trackId": track_id},
114 )
115
116 async def play(self, playlist_id: str) -> None:
117 await self._http.execute(
118 "mutation PlaySavedPlaylist($playlistId: String!) "
119 "{ playSavedPlaylist(playlistId: $playlistId) }",
120 {"playlistId": playlist_id},
121 )
122
123 # --- folders --------------------------------------------------------
124
125 async def folders(self) -> list[SavedPlaylistFolder]:
126 data = await self._http.execute(
127 f"query PlaylistFolders {{ playlistFolders {{ {_FOLDER_FIELDS} }} }}"
128 )
129 return [SavedPlaylistFolder.model_validate(f) for f in data.get("playlistFolders", [])]
130
131 async def create_folder(self, name: str) -> SavedPlaylistFolder:
132 data = await self._http.execute(
133 "mutation CreatePlaylistFolder($name: String!) "
134 f"{{ createPlaylistFolder(name: $name) {{ {_FOLDER_FIELDS} }} }}",
135 {"name": name},
136 )
137 return SavedPlaylistFolder.model_validate(data["createPlaylistFolder"])
138
139 async def delete_folder(self, id: str) -> None:
140 await self._http.execute(
141 "mutation DeletePlaylistFolder($id: String!) { deletePlaylistFolder(id: $id) }",
142 {"id": id},
143 )