diff --git a/pyproject.toml b/pyproject.toml index 9d68dca..9b37df3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [project] name = "oxidespotify" -version = "0.2.2" +version = "1.0.0" description = "Client for Spotify API" readme = "README.md" authors = [{ name = "Miwory", email = "miwory.uwu@gmail.com" }] requires-python = ">=3.14" -dependencies = ["oxidehttp>=1.0.3,<=2.0.0", "pydantic>=2.12,<=2.13"] +dependencies = ["oxidehttp>=1.4.0,<=2.0.0", "pydantic>=2.12,<=2.13"] [build-system] requires = ["uv_build>=0.9.2,<0.11.0"] diff --git a/src/oxidespotify/api.py b/src/oxidespotify/api.py index 352c04f..abe48bc 100644 --- a/src/oxidespotify/api.py +++ b/src/oxidespotify/api.py @@ -1,5 +1,5 @@ from collections.abc import Coroutine -from typing import Literal, NoReturn, overload +from typing import Literal, overload from oxidehttp.client import OxideHTTP from oxidehttp.schema import CachedResponse, Response @@ -33,6 +33,16 @@ class SpotifyAPIClient(OxideHTTP): return req + @overload + async def _process[T: BaseModel]( + self, req: Response | CachedResponse, schema: type[T] + ) -> T: ... + + @overload + async def _process[T: BaseModel]( + self, req: Response | CachedResponse, schema: None + ) -> None: ... + async def _process[T: BaseModel]( self, req: Response | CachedResponse, schema: type[T] | None ) -> T | None: @@ -51,47 +61,285 @@ class SpotifyAPIClient(OxideHTTP): if schema: return schema.model_validate(data) - async def get_album(self) -> NoReturn: - raise NotImplementedError + async def get_album( + self, + access_token: str, + album_id: str, + *, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.Album: + req = self.get( + f'/albums/{album_id}', + params=self.clean_dict({'id': album_id, 'market': market}), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def get_album_tracks(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.Album) - async def get_users_saved_albums(self) -> NoReturn: - raise NotImplementedError + async def get_album_tracks( + self, + access_token: str, + album_id: str, + *, + market: str | None = None, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.AlbumTracks: + req = self.get( + f'/albums/{album_id}/tracks', + params=self.clean_dict( + { + 'id': album_id, + 'market': market, + 'limit': limit, + 'offset': offset, + } + ), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def get_artist(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.AlbumTracks) - async def get_artists_albums(self) -> NoReturn: - raise NotImplementedError + async def get_users_saved_albums( + self, + access_token: str, + *, + limit: int = 20, + offset: int = 0, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.UserSavedAlbums: + req = self.get( + '/me/albums', + params=self.clean_dict( + { + 'market': market, + 'limit': limit, + 'offset': offset, + } + ), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def get_an_audiobook(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.UserSavedAlbums) - async def get_audiobook_chapters(self) -> NoReturn: - raise NotImplementedError + async def get_artist( + self, + access_token: str, + artist_id: str, + *, + cache_ttl: int | None = None, + ) -> s.Artist: + req = self.get( + f'/artists/{artist_id}', + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def get_users_saved_audiobooks(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.Artist) - async def get_a_chapter(self) -> NoReturn: - raise NotImplementedError + async def get_artists_albums( + self, + access_token: str, + artist_id: str, + *, + include_groups: list[ + Literal['album', 'single', 'appears_on', 'compilation'] + ] + | None = None, + market: str | None = None, + limit: int = 5, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.ArtistsAlbums: + req = self.get( + f'/artists/{artist_id}/albums', + params=self.clean_dict( + { + 'include_groups': ','.join(include_groups) + if include_groups + else None, + 'market': market, + 'limit': limit, + 'offset': offset, + } + ), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def get_episode(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.ArtistsAlbums) - async def get_users_saved_episodes(self) -> NoReturn: - raise NotImplementedError + async def get_an_audiobook( + self, + access_token: str, + audiobook_id: str, + *, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.AudioBook: + req = self.get( + f'/audiobooks/{audiobook_id}', + params=self.clean_dict({'id': audiobook_id, 'market': market}), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def save_item_to_library(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.AudioBook) - async def remove_item_from_library(self) -> NoReturn: - raise NotImplementedError + async def get_audiobook_chapters( + self, + access_token: str, + audiobook_id: str, + *, + market: str | None = None, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.AudiobookChapters: + req = self.get( + f'/audiobooks/{audiobook_id}/chapters', + params=self.clean_dict( + { + 'id': audiobook_id, + 'market': market, + 'limit': limit, + 'offset': offset, + } + ), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) - async def check_users_saved_items(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.AudiobookChapters) + + async def get_users_saved_audiobooks( + self, + access_token: str, + *, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.UserSavedAudiobooks: + req = self.get( + '/me/audiobooks', + params=self.clean_dict({'limit': limit, 'offset': offset}), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) + + req = await self._process_request(req) + return await self._process(req, s.UserSavedAudiobooks) + + async def get_a_chapter( + self, + access_token: str, + chapter_id: str, + *, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.Chapter: + req = self.get( + f'/chapters/{chapter_id}', + params=self.clean_dict({'id': chapter_id, 'market': market}), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) + + req = await self._process_request(req) + return await self._process(req, s.Chapter) + + async def get_episode( + self, + access_token: str, + episode_id: str, + *, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.Episode: + req = self.get( + f'/episodes/{episode_id}', + params=self.clean_dict({'id': episode_id, 'market': market}), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) + + req = await self._process_request(req) + return await self._process(req, s.Episode) + + async def get_users_saved_episodes( + self, + access_token: str, + *, + market: str | None = None, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.UserSavedEpisodes: + req = self.get( + '/me/episodes', + params=self.clean_dict( + {'market': market, 'limit': limit, 'offset': offset} + ), + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) + + req = await self._process_request(req) + return await self._process(req, s.UserSavedEpisodes) + + async def save_item_to_library( + self, access_token: str, uris: list[str] + ) -> None: + req = self.put( + '/me/library', + params={'uris': ','.join(uris)}, + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def remove_item_from_library( + self, access_token: str, uris: list[str] + ) -> None: + req = self.delete( + '/me/library', + params={'uris': ','.join(uris)}, + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def check_users_saved_items( + self, + access_token: str, + uris: list[str], + *, + cache_ttl: int | None = None, + ) -> s.CheckUserSavedItems: + req = self.get( + '/me/library/contains', + params={'uris': ','.join(uris)}, + headers=self._auth(access_token), + cache_ttl=cache_ttl, + ) + + req = await self._process_request(req) + return await self._process(req, s.CheckUserSavedItems) async def get_playback_state( self, @@ -118,22 +366,102 @@ class SpotifyAPIClient(OxideHTTP): req = await self._process_request(req) return await self._process(req, s.PlaybackState) - async def transfer_playback(self) -> NoReturn: - raise NotImplementedError + async def transfer_playback( + self, access_token: str, device_ids: str, *, play: bool = False + ) -> None: + req = self.put( + '/me/player', + params={'device_ids': device_ids, 'play': play}, + headers=self._auth(access_token), + ) - async def get_available_devices(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, None) - async def start_resume_playback(self) -> NoReturn: - raise NotImplementedError + async def get_available_devices( + self, access_token: str, *, cache_ttl: int + ) -> s.Devices: + req = self.get( + '/me/player/devices', + cache_ttl=cache_ttl, + headers=self._auth(access_token), + ) - async def pause_playback(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.Devices) - async def skip_to_next(self, access_token: str, device_id: str) -> None: + async def get_currently_playing_track( + self, + access_token: str, + *, + market: str | None = None, + additional_types: list[Literal['track', 'episode']] | None = None, + cache_ttl: int | None = None, + ) -> s.PlaybackState: + req = self.get( + '/me/player/currently-playing', + params=self.clean_dict( + { + 'market': market, + 'additional_types': ','.join(additional_types) + if additional_types + else None, + } + ), + cache_ttl=cache_ttl, + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.PlaybackState) + + async def start_resume_playback( + self, + access_token: str, + *, + device_id: str | None = None, + context_uri: str | None = None, + uris: list[str] | None = None, + offset_position: int | None = None, + position_ms: int | None = None, + ) -> None: + req = self.put( + '/me/player/play', + params=self.clean_dict( + { + 'device_id': device_id, + 'context_uri': context_uri, + 'uris': ','.join(uris) if uris else None, + 'offset': {'position': offset_position} + if offset_position + else None, + 'position_ms': position_ms, + } + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def pause_playback( + self, access_token: str, *, device_id: str + ) -> None: + req = self.put( + '/me/player/pause', + params=self.clean_dict({'device_id': device_id}), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def skip_to_next( + self, access_token: str, *, device_id: str | None = None + ) -> None: req = self.post( '/me/player/next', - params={'device_id': device_id}, + params=self.clean_dict({'device_id': device_id}), headers=self._auth(access_token), ) @@ -141,28 +469,84 @@ class SpotifyAPIClient(OxideHTTP): return await self._process(req, None) async def skip_to_previous( - self, access_token: str, device_id: str + self, access_token: str, *, device_id: str | None = None ) -> None: req = self.post( '/me/player/previous', - params={'device_id': device_id}, + params=self.clean_dict({'device_id': device_id}), headers=self._auth(access_token), ) req = await self._process_request(req) return await self._process(req, None) - async def seek_to_position(self) -> NoReturn: - raise NotImplementedError + async def seek_to_position( + self, + access_token: str, + position_ms: int, + *, + device_id: str | None = None, + ) -> None: + req = self.put( + '/me/player/seek', + params=self.clean_dict( + {'position_ms': position_ms, 'device_id': device_id} + ), + headers=self._auth(access_token), + ) - async def set_repeat_mode(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, None) - async def set_playback_volume(self) -> NoReturn: - raise NotImplementedError + async def set_repeat_mode( + self, + access_token: str, + state: Literal['track', 'context', 'off'], + *, + device_id: str | None = None, + ) -> None: + req = self.put( + '/me/player/repeat', + params=self.clean_dict({'state': state, 'device_id': device_id}), + headers=self._auth(access_token), + ) - async def toggle_playback_shuffle(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, None) + + async def set_playback_volume( + self, + access_token: str, + volume_percent: int, + *, + device_id: str | None = None, + ) -> None: + req = self.put( + '/me/player/volume', + params=self.clean_dict( + {'volume_percent': volume_percent, 'device_id': device_id} + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def toggle_playback_shuffle( + self, + access_token: str, + state: Literal['true', 'false'], + *, + device_id: str | None = None, + ) -> None: + req = self.put( + '/me/player/shuffle', + params=self.clean_dict({'state': state, 'device_id': device_id}), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) async def get_recently_played_tracks( self, @@ -172,7 +556,7 @@ class SpotifyAPIClient(OxideHTTP): after: int | None = None, before: int | None = None, cache_ttl: int | None = None, - ) -> s.RecentlyPlayed: + ) -> s.RecentlyPlayedTracks: req = self.get( '/me/player/recently-played', cache_ttl=cache_ttl, @@ -183,59 +567,251 @@ class SpotifyAPIClient(OxideHTTP): ) req = await self._process_request(req) - res = await self._process(req, s.RecentlyPlayed) + return await self._process(req, s.RecentlyPlayedTracks) - if res is None: - raise s.InternalError(status_code=0, error="Can't be here") - - return res - - async def get_the_users_queue(self) -> NoReturn: - raise NotImplementedError - - async def add_item_to_playback_queue( - self, access_token: str, uri: str, device_id: str - ) -> None: - req = self.post( + async def get_the_users_queue( + self, access_token: str, *, cache_ttl: int | None = None + ) -> s.UsersQueue: + req = self.get( '/me/player/queue', - params={'uri': uri, 'device_id': device_id}, + cache_ttl=cache_ttl, headers=self._auth(access_token), ) req = await self._process_request(req) - res = await self._process(req, None) + return await self._process(req, s.UsersQueue) - return res + async def add_item_to_playback_queue( + self, access_token: str, uri: str, *, device_id: str | None = None + ) -> None: + req = self.post( + '/me/player/queue', + params=self.clean_dict({'uri': uri, 'device_id': device_id}), + headers=self._auth(access_token), + ) - async def get_playlist(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, None) - async def change_playlist_details(self) -> NoReturn: - raise NotImplementedError + async def get_playlist( + self, + access_token: str, + playlist_id: str, + *, + market: str | None = None, + additional_types: list[Literal['track', 'episode']] | None = None, + cache_ttl: int | None = None, + ) -> s.Playlist: + req = self.get( + f'/playlists/{playlist_id}', + cache_ttl=cache_ttl, + params=self.clean_dict( + {'market': market, 'additional_types': additional_types} + ), + headers=self._auth(access_token), + ) - async def get_playlist_items(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.Playlist) - async def update_playlist_items(self) -> NoReturn: - raise NotImplementedError + async def change_playlist_details( + self, + access_token: str, + playlist_id: str, + *, + name: str | None = None, + public: Literal['true', 'false'] | None = None, + collaborative: Literal['true', 'false'] | None = None, + description: str | None = None, + ) -> None: + req = self.put( + f'/playlists/{playlist_id}', + json=self.clean_dict( + { + 'name': name, + 'public': public, + 'collaborative': collaborative, + 'description': description, + } + ), + headers=self._auth(access_token), + ) - async def add_items_to_playlist(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, None) - async def remove_playlist_items(self) -> NoReturn: - raise NotImplementedError + async def get_playlist_items( + self, + access_token: str, + playlist_id: str, + *, + market: str | None = None, + limit: int = 10, + offset: int = 0, + additional_types: list[Literal['track', 'episode']] | None = None, + cache_ttl: int | None = None, + ) -> s.PlaylistTracks: + req = self.get( + f'/playlists/{playlist_id}/tracks', + cache_ttl=cache_ttl, + params=self.clean_dict( + { + 'market': market, + 'limit': limit, + 'offset': offset, + 'additional_types': ','.join(additional_types) + if additional_types + else None, + } + ), + headers=self._auth(access_token), + ) - async def get_current_users_playlists(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.PlaylistTracks) - async def create_playlist(self) -> NoReturn: - raise NotImplementedError + async def update_playlist_items( + self, + access_token: str, + playlist_id: str, + uris: list[str], + *, + range_start: int | None = None, + insert_before: int | None = None, + range_length: int | None = None, + snapshot_id: str | None = None, + ) -> None: + req = self.put( + f'/playlists/{playlist_id}/tracks', + json=self.clean_dict( + { + 'uris': ','.join(uris), + 'range_start': range_start, + 'insert_before': insert_before, + 'range_length': range_length, + 'snapshot_id': snapshot_id, + } + ), + headers=self._auth(access_token), + ) - async def get_playlist_cover_image(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, None) - async def add_custom_playlist_cover_image(self) -> NoReturn: - raise NotImplementedError + async def add_items_to_playlist( + self, + access_token: str, + playlist_id: str, + uris: list[str], + *, + position: int | None = None, + ) -> None: + req = self.post( + f'/playlists/{playlist_id}/tracks', + json=self.clean_dict( + {'uris': ','.join(uris), 'position': position} + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def remove_playlist_items( + self, + access_token: str, + playlist_id: str, + items: list[str], + *, + snapshot_id: str | None = None, + ) -> None: + req = self.delete( + f'/playlists/{playlist_id}/tracks', + json=self.clean_dict( + { + 'items': [{'uri': item} for item in items], + 'snapshot_id': snapshot_id, + } + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, None) + + async def get_current_users_playlists( + self, + access_token: str, + *, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.UserPlaylists: + req = self.get( + '/me/playlists', + cache_ttl=cache_ttl, + params=self.clean_dict({'limit': limit, 'offset': offset}), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.UserPlaylists) + + async def create_playlist( + self, + access_token: str, + name: str, + *, + public: Literal['true', 'false'] | None = None, + collaborative: Literal['true', 'false'] | None = None, + description: str | None = None, + ) -> s.Playlist: + req = self.post( + '/me/playlists', + json=self.clean_dict( + { + 'name': name, + 'public': public, + 'collaborative': collaborative, + 'description': description, + } + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.Playlist) + + async def get_playlist_cover_image( + self, + access_token: str, + playlist_id: str, + *, + cache_ttl: int | None = None, + ) -> s.PlaylistImages: + req = self.get( + f'/playlists/{playlist_id}/images', + cache_ttl=cache_ttl, + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.PlaylistImages) + + async def add_custom_playlist_cover_image( + self, access_token: str, playlist_id: str, base64_image: str + ) -> None: + headers = { + 'Content-Type': 'image/jpeg', + } | self._auth(access_token) + req = self.put( + f'/playlists/{playlist_id}/images', + body_text=base64_image, + headers=headers, + ) + + req = await self._process_request(req) + return await self._process(req, None) @overload async def search_for_item( @@ -391,21 +967,65 @@ class SpotifyAPIClient(OxideHTTP): ) req = await self._process_request(req) - res = await self._process(req, target_model) + return await self._process(req, target_model) - if res is None: - raise s.InternalError(status_code=0, error="Can't be here") + async def get_show( + self, + access_token: str, + show_id: str, + *, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.Show: + req = self.get( + f'/shows/{show_id}', + cache_ttl=cache_ttl, + params=self.clean_dict({'market': market}), + headers=self._auth(access_token), + ) - return res + req = await self._process_request(req) + return await self._process(req, s.Show) - async def get_show(self) -> NoReturn: - raise NotImplementedError + async def get_show_episodes( + self, + access_token: str, + show_id: str, + *, + market: str | None = None, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.ShowEpisodes: + req = self.get( + f'/shows/{show_id}/episodes', + cache_ttl=cache_ttl, + params=self.clean_dict( + {'market': market, 'limit': limit, 'offset': offset} + ), + headers=self._auth(access_token), + ) - async def get_show_episodes(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.ShowEpisodes) - async def get_users_saved_shows(self) -> NoReturn: - raise NotImplementedError + async def get_users_saved_shows( + self, + access_token: str, + *, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.SavedShows: + req = self.get( + '/me/shows', + cache_ttl=cache_ttl, + params=self.clean_dict({'limit': limit, 'offset': offset}), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.SavedShows) async def get_track( self, @@ -423,15 +1043,28 @@ class SpotifyAPIClient(OxideHTTP): ) req = await self._process_request(req) - res = await self._process(req, s.Track) + return await self._process(req, s.Track) - if res is None: - raise s.InternalError(status_code=0, error="Can't be here") + async def get_users_saved_tracks( + self, + access_token: str, + *, + market: str, + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.SavedTracks: + req = self.get( + '/me/tracks', + cache_ttl=cache_ttl, + params=self.clean_dict( + {'market': market, 'limit': limit, 'offset': offset} + ), + headers=self._auth(access_token), + ) - return res - - async def get_users_saved_tracks(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + return await self._process(req, s.SavedTracks) async def get_current_users_profile( self, access_token: str, *, cache_ttl: int | None = None @@ -443,15 +1076,80 @@ class SpotifyAPIClient(OxideHTTP): ) req = await self._process_request(req) - res = await self._process(req, s.UserProfile) + return await self._process(req, s.UserProfile) - if res is None: - raise s.InternalError(status_code=0, error="Can' be here") + @overload + async def get_users_top_items( + self, + access_token: str, + item_type: Literal['artists'], + *, + time_range: Literal[ + 'long_term', 'medium_term', 'short_term' + ] = 'medium_term', + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.UsersTopItemsArtists: ... - return res + @overload + async def get_users_top_items( + self, + access_token: str, + item_type: Literal['track'], + *, + time_range: Literal[ + 'long_term', 'medium_term', 'short_term' + ] = 'medium_term', + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.UsersTopItemsTracks: ... - async def get_users_top_items(self) -> NoReturn: - raise NotImplementedError + async def get_users_top_items( + self, + access_token: str, + item_type: Literal['artists', 'track'], + *, + time_range: Literal[ + 'long_term', 'medium_term', 'short_term' + ] = 'medium_term', + limit: int = 20, + offset: int = 0, + cache_ttl: int | None = None, + ) -> s.UsersTopItemsTracks | s.UsersTopItemsArtists: + req = self.get( + f'/me/top/{item_type}', + cache_ttl=cache_ttl, + params=self.clean_dict( + {'time_range': time_range, 'limit': limit, 'offset': offset} + ), + headers=self._auth(access_token), + ) - async def get_followed_artists(self) -> NoReturn: - raise NotImplementedError + req = await self._process_request(req) + + if item_type == 'artists': + return await self._process(req, s.UsersTopItemsArtists) + + return await self._process(req, s.UsersTopItemsTracks) + + async def get_followed_artists( + self, + access_token: str, + *, + after: str | None = None, + limit: int = 20, + cache_ttl: int | None = None, + ) -> s.FollowedArtists: + req = self.get( + '/me/following', + cache_ttl=cache_ttl, + params=self.clean_dict( + {'after': after, 'limit': limit, 'type': 'artist'} + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.FollowedArtists) diff --git a/src/oxidespotify/schema.py b/src/oxidespotify/schema.py index 31e8145..ca45ce8 100644 --- a/src/oxidespotify/schema.py +++ b/src/oxidespotify/schema.py @@ -1,7 +1,7 @@ from datetime import datetime from typing import Literal -from pydantic import BaseModel, HttpUrl, field_validator +from pydantic import BaseModel, HttpUrl, RootModel, field_validator class Error(Exception): @@ -62,6 +62,20 @@ class Image(BaseModel): return v +class Cursors(BaseModel): + after: int + before: int + + +class Paginated(BaseModel): + href: HttpUrl + limit: int + next: HttpUrl | None + offset: int = 0 + previous: HttpUrl | None + total: int + + class ExternalUrls(BaseModel): spotify: HttpUrl @@ -72,6 +86,16 @@ class ExternalIDs(BaseModel): upc: str | None = None +class Copyright(BaseModel): + text: str + type: Literal['C', 'P'] + + +class ResumePoint(BaseModel): + fully_played: bool + resume_position_ms: int + + class Artist(BaseModel): external_urls: ExternalUrls href: HttpUrl @@ -81,19 +105,28 @@ class Artist(BaseModel): uri: str -class Cursor(BaseModel): - after: int - before: int - - -class Context(BaseModel): - type: Literal['artist', 'playlist', 'album', 'show', 'collection'] - href: HttpUrl +class BaseTrack(BaseModel): + artists: list[Artist] + disc_number: int + duration_ms: int + explicit: bool external_urls: ExternalUrls + href: HttpUrl + id: str + is_playable: bool | None = None + restrictions: Restriction | None = None + name: str + track_number: int + type: Literal['track'] uri: str + is_local: bool -class Album(BaseModel): +class AlbumTracks(Paginated): + items: list[BaseTrack] + + +class BaseAlbum(BaseModel): album_type: Literal['album', 'single', 'compilation'] total_tracks: int external_urls: ExternalUrls @@ -109,103 +142,27 @@ class Album(BaseModel): artists: list[Artist] -class PlaylistOwner(BaseModel): - external_urls: ExternalUrls - href: HttpUrl - id: str - type: Literal['user'] - uri: str - display_name: str | None - - -class PlaylistTracks(BaseModel): - href: HttpUrl - total: int - - -class Playlist(BaseModel): - collaborative: bool - description: str - external_urls: ExternalUrls - href: HttpUrl - id: str - images: list[Image] - name: str - owner: PlaylistOwner - public: bool - snapshot_id: str - items: PlaylistTracks - type: Literal['playlist'] - uri: str - - -class Track(BaseModel): - album: Album - artists: list[Artist] - disc_number: int - duration_ms: int - explicit: bool +class Album(BaseAlbum): external_ids: ExternalIDs - external_urls: ExternalUrls - href: HttpUrl - id: str - is_playable: bool | None = None - restrictions: Restriction | None = None - name: str - track_number: int - type: Literal['track'] - uri: str - is_local: bool - - -class EpisodeResumePoint(BaseModel): - fully_played: bool - resume_position_ms: int - - -class Copyright(BaseModel): - text: str - type: Literal['C', 'P'] - - -class Show(BaseModel): copyrights: list[Copyright] - description: str - html_description: str - explicit: bool - external_ids: ExternalIDs | None = None - href: HttpUrl - id: str - images: list[Image] - is_externally_hosted: bool - languages: list[str] - media_type: str - name: str - type: Literal['show'] - uri: str - total_episodes: int + tracks: AlbumTracks -class Episode(BaseModel): - description: str - html_description: str - duration_ms: int - explicit: bool - external_urls: ExternalUrls - href: HttpUrl - id: str - images: list[Image] - is_externally_hosted: bool - is_playable: bool - languages: list[str] - name: str - release_date: datetime - release_date_precision: Literal['year', 'month', 'day'] - resume_point: EpisodeResumePoint - type: Literal['episode'] - uri: str - restrictions: Restriction | None = None - show: Show | None = None +class Track(BaseTrack): + album: BaseAlbum + + +class UserSavedAlbum(BaseModel): + added_at: datetime + album: Album + + +class UserSavedAlbums(Paginated): + items: list[UserSavedAlbum] + + +class ArtistsAlbums(Paginated): + items: list[Album] class AudioBookAuthor(BaseModel): @@ -236,17 +193,79 @@ class AudioBook(BaseModel): total_chapters: int -class PlaybackStateActions(BaseModel): - interrupting_playback: bool | None = None - pausing: bool | None = None - resuming: bool | None = None - seeking: bool | None = None - skipping_next: bool | None = None - skipping_prev: bool | None = None - toggling_repeat_context: bool | None = None - toggling_shuffle: bool | None = None - toggling_repeat_track: bool | None = None - transferring_playback: bool | None = None +class AudioBookChapter(BaseModel): + chapter_number: int + description: str + html_description: str + duration_ms: int + explicit: bool + external_urls: ExternalUrls + href: HttpUrl + id: str + images: list[Image] + is_playable: bool + languages: list[str] + name: str + release_date: datetime + release_date_precision: Literal['year', 'month', 'day'] + resume_point: ResumePoint + type: Literal['episode'] + uri: str + restrictions: Restriction + + +class AudiobookChapters(BaseModel): + items: list[AudioBookChapter] + + +class UserSavedAudiobooks(Paginated): + items: list[AudioBook] + + +class Chapter(AudioBookChapter): + audiobook: AudioBook + + +class Show(BaseModel): + copyrights: list[Copyright] + description: str + html_description: str + explicit: bool + external_ids: ExternalIDs | None = None + href: HttpUrl + id: str + images: list[Image] + is_externally_hosted: bool + languages: list[str] + media_type: str + name: str + type: Literal['show'] + uri: str + total_episodes: int + + +class Episode(AudioBookChapter): + show: Show + + +class UsedSavedEpisode(BaseModel): + added_at: datetime + episode: Episode + + +class UserSavedEpisodes(Paginated): + items: list[UsedSavedEpisode] + + +class CheckUserSavedItems(RootModel[list[bool]]): + pass + + +class Context(BaseModel): + type: Literal['artist', 'playlist', 'album', 'show', 'collection'] + href: HttpUrl + external_urls: ExternalUrls + uri: str class PlaybackState(BaseModel): @@ -259,35 +278,94 @@ class PlaybackState(BaseModel): is_playing: bool item: Track | Episode | None currently_playing_type: Literal['track', 'episode', 'ad', 'unknown'] - # TODO: allows has 'disallows' inside? not sure about that - # actions: PlaybackStateActions -class PlayHistoryObject(BaseModel): +class Devices(BaseModel): + devices: list[Device] + + +class RecentlyPlayedTrack(BaseModel): track: Track played_at: datetime - context: Context | None = None + context: Context -class RecentlyPlayed(BaseModel): +class RecentlyPlayedTracks(Paginated): + cursors: Cursors + items: list[RecentlyPlayedTrack] + + +class UsersQueue(BaseModel): + currently_playing: Track | Episode | None + queue: list[Track | Episode] + + +class PlaylistOwner(BaseModel): + external_urls: ExternalUrls href: HttpUrl - limit: int - next: HttpUrl | None - cursors: Cursor - total: int | None = None - items: list[PlayHistoryObject] + id: str + type: Literal['user'] + uri: str + display_name: str | None -class SearchData(BaseModel): +class PlaylistTrack(BaseModel): + added_at: datetime + added_by: PlaylistOwner + is_local: bool + item: Track | Episode + + +class PlaylistTracks(Paginated): + href: HttpUrl + item: PlaylistTrack + + +class SimplifiedPlaylistTracks(Paginated): href: HttpUrl - limit: int - next: HttpUrl | None - offset: int - previous: int | None total: int -class SearchAlbumData(SearchData): +class Playlist(BaseModel): + type: Literal['playlist'] + uri: str + collaborative: bool + description: str + external_urls: ExternalUrls + href: HttpUrl + id: str + images: list[Image] + name: str + owner: PlaylistOwner + public: bool + snapshot_id: str + items: PlaylistTracks | None = None + + +class SimplifiedPlaylist(BaseModel): + type: Literal['playlist'] + uri: str + collaborative: bool + external_urls: ExternalUrls + href: HttpUrl + id: str + images: list[Image] + name: str + owner: PlaylistOwner + public: bool + snapshot_id: str + items: SimplifiedPlaylistTracks + + +class UserPlaylists(Paginated): + items: list[Playlist] + + +class PlaylistImages(RootModel[list[Image]]): + pass + + +class SearchAlbumData(Paginated): items: list[Album] @@ -295,7 +373,7 @@ class SearchAlbum(BaseModel): albums: SearchAlbumData -class SearchArtistData(SearchData): +class SearchArtistData(Paginated): items: list[Artist] @@ -303,7 +381,7 @@ class SearchArtist(BaseModel): artists: SearchArtistData -class SearchPlaylistData(SearchData): +class SearchPlaylistData(Paginated): items: list[Playlist | None] @@ -311,7 +389,7 @@ class SearchPlaylist(BaseModel): playlists: SearchPlaylistData -class SearchTrackData(SearchData): +class SearchTrackData(Paginated): items: list[Track] @@ -319,7 +397,7 @@ class SearchTrack(BaseModel): tracks: SearchTrackData -class SearchShowData(SearchData): +class SearchShowData(Paginated): items: list[Show] @@ -327,7 +405,7 @@ class SearchShow(BaseModel): shows: SearchShowData -class SearchEpisodeData(SearchData): +class SearchEpisodeData(Paginated): items: list[Episode] @@ -335,7 +413,7 @@ class SearchEpisode(BaseModel): episodes: SearchEpisodeData -class SearchAudiobookData(SearchData): +class SearchAudiobookData(Paginated): items: list[AudioBook] @@ -343,6 +421,28 @@ class SearchAudiobook(BaseModel): audiobooks: SearchAudiobookData +class ShowEpisodes(Paginated): + items: list[AudioBookChapter] + + +class SavedShow(BaseModel): + added_at: datetime + show: Show + + +class SavedShows(Paginated): + items: list[SavedShow] + + +class SavedTrack(BaseModel): + added_at: datetime + track: Track + + +class SavedTracks(Paginated): + items: list[SavedTrack] + + class UserProfile(BaseModel): id: str display_name: str @@ -351,3 +451,19 @@ class UserProfile(BaseModel): images: list[Image] type: Literal['user'] uri: str + + +class UsersTopItemsTracks(Paginated): + items: list[Track] + + +class UsersTopItemsArtists(Paginated): + items: list[Artist] + + +class FollowedArtistsData(Paginated): + items: list[Artist] + + +class FollowedArtists(BaseModel): + artists: FollowedArtistsData