from datetime import datetime
from typing import List, Optional, Union, Iterable
from platypush.common.spotify import SpotifyMixin
from platypush.message.response import Response
from platypush.plugins import action
from platypush.plugins.media import PlayerState
from platypush.plugins.music import MusicPlugin
from platypush.schemas.spotify import (
SpotifyDeviceSchema,
SpotifyStatusSchema,
SpotifyTrackSchema,
SpotifyHistoryItemSchema,
SpotifyPlaylistSchema,
SpotifyAlbumSchema,
SpotifyEpisodeSchema,
SpotifyShowSchema,
SpotifyArtistSchema,
)
[docs]class MusicSpotifyPlugin(MusicPlugin, SpotifyMixin):
"""
Plugin to interact with the user's Spotify library and players.
In order to use this plugin to interact with your Spotify account you need to register a new app on the Spotify
developers website, whitelist the callback URL of your Platypush host and authorize the app to your account:
- Create a developer app on https://developer.spotify.com.
- Get the app's ``client_id`` and ``client_secret``.
- Whitelist the authorization callback URL on the Platypush machine, usually in the form
``http(s)://your-platypush-hostname-or-local-ip:8008/spotify/auth_callback`` (you need the
``http`` Platypush backend to be enabled).
- You can then authorize the app by opening the following URL in a browser:
``https://accounts.spotify.com/authorize?client_id=<client_id>&response_type=code&redirect_uri=http(s)://your-platypush-hostname-or-local-ip:8008/spotify/auth_callback&scope=<comma-separated-list-of-scopes>&state=<some-random-string>``.
This is the list of scopes required for full plugin functionalities:
- ``user-read-playback-state``
- ``user-modify-playback-state``
- ``user-read-currently-playing``
- ``user-read-recently-played``
- ``app-remote-control``
- ``streaming``
- ``playlist-modify-public``
- ``playlist-modify-private``
- ``playlist-read-private``
- ``playlist-read-collaborative``
- ``user-library-modify``
- ``user-library-read``
Alternatively, you can call any of the methods from this plugin over HTTP API, and the full authorization URL should
be printed on the application logs/stdout.
"""
[docs] def __init__(
self,
client_id: Optional[str] = None,
client_secret: Optional[str] = None,
**kwargs,
):
MusicPlugin.__init__(self, **kwargs)
SpotifyMixin.__init__(
self, client_id=client_id, client_secret=client_secret, **kwargs
)
self._players_by_id = {}
self._players_by_name = {}
# Playlist ID -> snapshot ID and tracks cache
self._playlist_snapshots = {}
def _get_device(self, device: str):
dev = self._players_by_id.get(device, self._players_by_name.get(device))
if not dev:
self.get_devices()
dev = self._players_by_id.get(device, self._players_by_name.get(device))
assert dev, f'No such device: {device}'
return dev
@staticmethod
def _parse_datetime(
dt: Optional[Union[str, datetime, int, float]]
) -> Optional[datetime]:
if isinstance(dt, str):
try:
dt = float(dt)
except (ValueError, TypeError):
return datetime.fromisoformat(dt)
if isinstance(dt, (int, float)):
return datetime.fromtimestamp(dt)
return dt
[docs] @action
def get_devices(self) -> List[dict]:
"""
Get the list of players associated to the Spotify account.
:return: .. schema:: spotify.SpotifyDeviceSchema(many=True)
"""
devices = self.spotify_user_call('/v1/me/player/devices').get('devices', [])
self._players_by_id = {
**self._players_by_id,
**{dev['id']: dev for dev in devices},
}
self._players_by_name = {
**self._players_by_name,
**{dev['name']: dev for dev in devices},
}
return SpotifyDeviceSchema().dump(devices, many=True)
[docs] @action
def set_volume(self, volume: int, device: Optional[str] = None):
"""
Set the playback volume on a device.
:param volume: Target volume as a percentage between 0 and 100.
:param device: Device ID or name. If none is specified then the currently active device will be used.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/volume',
method='put',
params={
'volume_percent': volume,
**({'device_id': device} if device else {}),
},
)
def _get_volume(self, device: Optional[str] = None) -> Optional[int]:
if device:
return self._get_device(device).get('volume')
return self.status.output.get('volume')
[docs] @action
def volup(self, delta: int = 5, device: Optional[str] = None):
"""
Set the volume up by a certain delta.
:param delta: Increase the volume by this percentage amount (between 0 and 100).
:param device: Device ID or name. If none is specified then the currently active device will be used.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/volume',
params={
'volume_percent': min(100, (self._get_volume() or 0) + delta),
**({'device_id': device} if device else {}),
},
)
[docs] @action
def voldown(self, delta: int = 5, device: Optional[str] = None):
"""
Set the volume down by a certain delta.
:param delta: Decrease the volume by this percentage amount (between 0 and 100).
:param device: Device ID or name. If none is specified then the currently active device will be used.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/volume',
params={
'volume_percent': max(0, (self._get_volume() or 0) - delta),
**({'device_id': device} if device else {}),
},
)
[docs] @action
def play(self, resource: Optional[str] = None, device: Optional[str] = None):
"""
Change the playback state of a device to ``PLAY`` or start playing a specific resource.
:param resource: Resource to play, in Spotify URI format (e.g. ``spotify:track:xxxxxxxxxxxxxxxxxxxxxx``).
If none is specified then the method will change the playback state to ``PLAY``.
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/play',
method='put',
json={'uris': [resource]} if resource else {},
params={
**({'device_id': device} if device else {}),
},
)
[docs] @action
def pause(self, device: Optional[str] = None):
"""
Toggle paused state.
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
# noinspection PyUnresolvedReferences
status = self.status().output
state = (
'play'
if status.get('device_id') != device
or status.get('state') != PlayerState.PLAY.value
else 'pause'
)
self.spotify_user_call(
f'/v1/me/player/{state}',
method='put',
params={
**({'device_id': device} if device else {}),
},
)
[docs] @action
def pause_if_playing(self):
"""
Pause playback only if it's playing
"""
# noinspection PyUnresolvedReferences
status = self.status().output
if status.get('state') == PlayerState.PLAY.value:
self.spotify_user_call(
'/v1/me/player/pause',
method='put',
)
[docs] @action
def play_if_paused(self, device: Optional[str] = None):
"""
Play only if it's paused (resume)
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
# noinspection PyUnresolvedReferences
status = self.status().output
if status.get('state') != PlayerState.PLAY.value:
self.spotify_user_call(
'/v1/me/player/play',
method='put',
params={
**({'device_id': device} if device else {}),
},
)
[docs] @action
def play_if_paused_or_stopped(self):
"""
Alias for :meth:`.play_if_paused`.
"""
return self.play_if_paused()
[docs] @action
def stop(self, **kwargs):
"""
This method is actually just an alias to :meth:`.stop`, since Spotify manages clearing playback sessions
automatically after a while for paused devices.
"""
return self.pause(**kwargs)
[docs] @action
def start_or_transfer_playback(self, device: str):
"""
Start or transfer playback to the device specified.
:param device: Device ID or name.
"""
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player',
method='put',
json={
'device_ids': [device],
},
)
[docs] @action
def next(self, device: Optional[str] = None, **kwargs):
"""
Skip to the next track.
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/next',
method='post',
params={
**({'device_id': device} if device else {}),
},
)
[docs] @action
def previous(self, device: Optional[str] = None, **kwargs):
"""
Skip to the next track.
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/previous',
method='post',
params={
**({'device_id': device} if device else {}),
},
)
[docs] @action
def seek(self, position: float, device: Optional[str] = None, **kwargs):
"""
Set the cursor to the specified position in the track.
:param position: Position in seconds.
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/seek',
method='put',
params={
'position_ms': int(position * 1000),
**({'device_id': device} if device else {}),
},
)
[docs] @action
def repeat(self, value: Optional[bool] = None, device: Optional[str] = None):
"""
Set or toggle repeat mode.
:param value: If set, set the repeat state this value (true/false). Default: None (toggle current state).
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
if value is None:
# noinspection PyUnresolvedReferences
status = self.status().output
state = (
'context'
if status.get('device_id') != device or not status.get('repeat')
else 'off'
)
else:
state = value is True
self.spotify_user_call(
'/v1/me/player/repeat',
method='put',
params={
'state': 'context' if state else 'off',
**({'device_id': device} if device else {}),
},
)
[docs] @action
def random(self, value: Optional[bool] = None, device: Optional[str] = None):
"""
Set or toggle random/shuffle mode.
:param value: If set, set the shuffle state this value (true/false). Default: None (toggle current state).
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
if value is None:
# noinspection PyUnresolvedReferences
status = self.status().output
state = bool(status.get('device_id') != device or not status.get('random'))
else:
state = value is True
self.spotify_user_call(
'/v1/me/player/shuffle',
method='put',
params={
'state': state,
**({'device_id': device} if device else {}),
},
)
[docs] @action
def history(
self,
limit: int = 20,
before: Optional[Union[datetime, str, int]] = None,
after: Optional[Union[datetime, str, int]] = None,
):
"""
Get a list of recently played track on the account.
:param limit: Maximum number of tracks to be retrieved (default: 20, max: 50).
:param before: Retrieve only the tracks played before this timestamp, specified as a UNIX timestamp, a datetime
object or an ISO datetime string. If ``before`` is set then ``after`` cannot be set.
:param after: Retrieve only the tracks played after this timestamp, specified as a UNIX timestamp, a datetime
object or an ISO datetime string. If ``after`` is set then ``before`` cannot be set.
:return:
"""
before = self._parse_datetime(before)
after = self._parse_datetime(after)
assert not (before and after), 'before and after cannot both be set'
results = self._spotify_paginate_results(
'/v1/me/player/recently-played',
limit=limit,
params={
'limit': min(limit, 50),
**({'before': before} if before else {}),
**({'after': after} if after else {}),
},
)
return SpotifyHistoryItemSchema().dump(
[
{
**item.pop('track'),
**item,
}
for item in results
],
many=True,
)
[docs] @action
def add(self, resource: str, device: Optional[str] = None, **kwargs):
"""
Add a Spotify resource (track, or episode) to the playing queue.
:param resource: Spotify resource URI.
:param device: Device ID or name. If none is specified then the action will target the currently active device.
"""
if device:
device = self._get_device(device)['id']
self.spotify_user_call(
'/v1/me/player/queue',
method='post',
params={
'uri': resource,
**({'device_id': device} if device else {}),
},
)
@action
def clear(self, **kwargs):
pass
[docs] @action
def status(self, **kwargs) -> dict:
"""
Get the status of the currently active player.
:return: .. schema:: spotify.SpotifyStatusSchema
"""
status = self.spotify_user_call('/v1/me/player')
if not status:
return {
'state': PlayerState.STOP.value,
}
return SpotifyStatusSchema().dump(status)
[docs] @action
def current_track(self, **kwargs) -> dict:
"""
Get the track currently playing.
:return: .. schema:: spotify.SpotifyTrackSchema
"""
status = self.spotify_user_call('/v1/me/player')
empty_response = Response(output={})
if not status:
# noinspection PyTypeChecker
return empty_response
track = status.get('item', {})
if not track:
# noinspection PyTypeChecker
return empty_response
return SpotifyTrackSchema().dump(track)
[docs] @action
def get_playlists(
self, limit: int = 1000, offset: int = 0, user: Optional[str] = None
):
"""
Get the user's playlists.
:param limit: Maximum number of results (default: 1000).
:param offset: Return results starting from this index (default: 0).
:param user: Return the playlist owned by a specific user ID (default: currently logged in user).
:return: .. schema:: spotify.SpotifyPlaylistSchema
"""
playlists = self._spotify_paginate_results(
f'/v1/{"users/" + user if user else "me"}/playlists',
limit=limit,
offset=offset,
)
return SpotifyPlaylistSchema().dump(playlists, many=True)
def _get_playlist(self, playlist: str) -> dict:
playlists = self.get_playlists().output
playlists = [
pl
for pl in playlists
if (pl['id'] == playlist or pl['uri'] == playlist or pl['name'] == playlist)
]
assert playlists, f'No such playlist ID, URI or name: {playlist}'
return playlists[0]
def _get_playlist_tracks_from_cache(
self, id: str, snapshot_id: str, limit: Optional[int] = None, offset: int = 0
) -> Optional[Iterable]:
snapshot = self._playlist_snapshots.get(id)
if (
not snapshot
or snapshot['snapshot_id'] != snapshot_id
or (limit is None and snapshot['limit'] is not None)
):
return
if limit is not None and snapshot['limit'] is not None:
stored_range = (snapshot['limit'], snapshot['limit'] + snapshot['offset'])
requested_range = (limit, limit + offset)
if (
requested_range[0] < stored_range[0]
or requested_range[1] > stored_range[1]
):
return
return snapshot['tracks']
def _cache_playlist_data(
self,
id: str,
snapshot_id: str,
tracks: Iterable[dict],
limit: Optional[int] = None,
offset: int = 0,
**_,
):
self._playlist_snapshots[id] = {
'id': id,
'tracks': tracks,
'snapshot_id': snapshot_id,
'limit': limit,
'offset': offset,
}
[docs] @action
def get_playlist(
self,
playlist: str,
with_tracks: bool = True,
limit: Optional[int] = None,
offset: int = 0,
):
"""
Get a playlist content.
:param playlist: Playlist name, ID or URI.
:param with_tracks: Return also the playlist tracks (default: false, return only the metadata).
:param limit: If ``with_tracks`` is True, retrieve this maximum amount of tracks
(default: None, get all tracks).
:param offset: If ``with_tracks`` is True, retrieve tracks starting from this index (default: 0).
:return: .. schema:: spotify.SpotifyPlaylistSchema
"""
playlist = self._get_playlist(playlist)
if with_tracks:
playlist['tracks'] = self._get_playlist_tracks_from_cache(
playlist['id'],
snapshot_id=playlist['snapshot_id'],
limit=limit,
offset=offset,
)
if playlist['tracks'] is None:
playlist['tracks'] = [
{
**track,
'track': {
**track['track'],
'position': offset + i + 1,
},
}
for i, track in enumerate(
self._spotify_paginate_results(
f'/v1/playlists/{playlist["id"]}/tracks',
limit=limit,
offset=offset,
)
)
]
self._cache_playlist_data(**playlist, limit=limit, offset=offset)
return SpotifyPlaylistSchema().dump(playlist)
[docs] @action
def add_to_playlist(
self,
playlist: str,
resources: Union[str, Iterable[str]],
position: Optional[int] = None,
):
"""
Add one or more items to a playlist.
:param playlist: Playlist name, ID or URI.
:param resources: URI(s) of the resource(s) to be added.
:param position: At what (1-based) position the tracks should be inserted (default: append to the end).
"""
playlist = self._get_playlist(playlist)
response = self.spotify_user_call(
f'/v1/playlists/{playlist["id"]}/tracks',
method='post',
params={
**({'position': position} if position is not None else {}),
},
json={
'uris': [
uri.strip()
for uri in (
resources.split(',')
if isinstance(resources, str)
else resources
)
]
},
)
snapshot_id = response.get('snapshot_id')
assert snapshot_id is not None, 'Could not save playlist'
[docs] @action
def remove_from_playlist(self, playlist: str, resources: Union[str, Iterable[str]]):
"""
Remove one or more items from a playlist.
:param playlist: Playlist name, ID or URI.
:param resources: URI(s) of the resource(s) to be removed. A maximum of 100 tracks can be provided at once.
"""
playlist = self._get_playlist(playlist)
response = self.spotify_user_call(
f'/v1/playlists/{playlist["id"]}/tracks',
method='delete',
json={
'tracks': [
{'uri': uri.strip()}
for uri in (
resources.split(',')
if isinstance(resources, str)
else resources
)
]
},
)
snapshot_id = response.get('snapshot_id')
assert snapshot_id is not None, 'Could not save playlist'
[docs] @action
def playlist_move(
self,
playlist: str,
from_pos: int,
to_pos: int,
range_length: int = 1,
resources: Optional[Union[str, Iterable[str]]] = None,
**_,
):
"""
Move or replace elements in a playlist.
:param playlist: Playlist name, ID or URI.
:param from_pos: Move tracks starting from this position (the first element has index 1).
:param to_pos: Move tracks to this position (1-based index).
:param range_length: Number of tracks to move (default: 1).
:param resources: If specified, then replace the items from `from_pos` to `from_pos+range_length` with the
specified set of Spotify URIs (it must be a collection with the same length as the range).
"""
playlist = self._get_playlist(playlist)
response = self.spotify_user_call(
f'/v1/playlists/{playlist["id"]}/tracks',
method='put',
json={
'range_start': int(from_pos) + 1,
'range_length': int(range_length),
'insert_before': int(to_pos) + 1,
**(
{
'uris': [
uri.strip()
for uri in (
resources.split(',')
if isinstance(resources, str)
else resources
)
]
}
if resources
else {}
),
},
)
snapshot_id = response.get('snapshot_id')
assert snapshot_id is not None, 'Could not save playlist'
# noinspection PyShadowingBuiltins
@staticmethod
def _make_filter(query: Union[str, dict], **filter) -> str:
if filter:
query = {
**({'any': query} if isinstance(query, str) else {}),
**filter,
}
if isinstance(query, str):
return query
q = query['any'] if 'any' in query else ''
for attr in ['artist', 'track', 'album', 'year']:
if attr in query:
q += f' {attr}:{query[attr]}'
return q.strip()
# noinspection PyShadowingBuiltins
[docs] @action
def search(
self,
query: Optional[Union[str, dict]] = None,
limit: int = 50,
offset: int = 0,
type: str = 'track',
**filter,
) -> Iterable[dict]:
"""
Search for tracks matching a certain criteria.
:param query: Search filter. It can either be a free-text or a structured query. In the latter case the
following fields are supported:
- ``any``: Search for anything that matches this text.
- ``uri``: Search the following Spotify ID/URI or list of IDs/URIs.
- ``artist``: Filter by artist.
- ``track``: Filter by track name.
- ``album``: Filter by album name.
- ``year``: Filter by year (dash-separated ranges are supported).
:param limit: Maximum number of results (default: 50).
:param offset: Return results starting from this index (default: 0).
:param type: Type of results to be returned. Supported: ``album``, ``artist``, ``playlist``, ``track``, ``show``
and ``episode`` (default: ``track``).
:param filter: Alternative key-value way of representing a structured query.
:return:
If ``type=track``:
.. schema:: spotify.SpotifyTrackSchema(many=True)
If ``type=album``:
.. schema:: spotify.SpotifyAlbumSchema(many=True)
If ``type=artist``:
.. schema:: spotify.SpotifyArtistSchema(many=True)
If ``type=playlist``:
.. schema:: spotify.SpotifyPlaylistSchema(many=True)
If ``type=episode``:
.. schema:: spotify.SpotifyEpisodeSchema(many=True)
If ``type=show``:
.. schema:: spotify.SpotifyShowSchema(many=True)
"""
uri = {
**(query if isinstance(query, dict) else {}),
**filter,
}.get('uri', [])
uris = uri.split(',') if isinstance(uri, str) else uri
params = (
{
'ids': ','.join([uri.split(':')[-1].strip() for uri in uris]),
}
if uris
else {
'q': self._make_filter(query, **filter),
'type': type,
}
)
response = self._spotify_paginate_results(
f'/v1/{type + "s" if uris else "search"}',
limit=limit,
offset=offset,
type=type,
params=params,
)
if type == 'track':
return sorted(
SpotifyTrackSchema(many=True).dump(response),
key=lambda track: (
track.get('artist'),
track.get('date'),
track.get('album'),
track.get('track'),
track.get('title'),
track.get('popularity'),
),
)
schema_class = None
if type == 'playlist':
schema_class = SpotifyPlaylistSchema
if type == 'album':
schema_class = SpotifyAlbumSchema
if type == 'artist':
schema_class = SpotifyArtistSchema
if type == 'episode':
schema_class = SpotifyEpisodeSchema
if type == 'show':
schema_class = SpotifyShowSchema
if schema_class:
return schema_class(many=True).dump(response)
return response
[docs] @action
def create_playlist(
self, name: str, description: Optional[str] = None, public: bool = False
):
"""
Create a playlist.
:param name: Playlist name.
:param description: Optional playlist description.
:param public: Whether the new playlist should be public
(default: False).
:return: .. schema:: spotify.SpotifyPlaylistSchema
"""
ret = self.spotify_user_call(
'/v1/users/me/playlists',
method='post',
json={
'name': name,
'description': description,
'public': public,
},
)
return SpotifyPlaylistSchema().dump(ret)
[docs] @action
def follow_playlist(self, playlist: str, public: bool = True):
"""
Follow a playlist.
:param playlist: Playlist name, ID or URI.
:param public: If True (default) then the playlist will appear in the user's list of public playlists, otherwise
it won't.
"""
playlist = self._get_playlist(playlist)
self.spotify_user_call(
f'/v1/playlists/{playlist["id"]}/followers',
method='put',
json={
'public': public,
},
)
[docs] @action
def unfollow_playlist(self, playlist: str):
"""
Unfollow a playlist.
:param playlist: Playlist name, ID or URI.
"""
playlist = self._get_playlist(playlist)
self.spotify_user_call(
f'/v1/playlists/{playlist["id"]}/followers',
method='delete',
)
@staticmethod
def _uris_to_id(*uris: str) -> Iterable[str]:
return [uri.split(':')[-1] for uri in uris]
[docs] @action
def get_albums(self, limit: int = 50, offset: int = 0) -> List[dict]:
"""
Get the list of albums saved by the user.
:param limit: Maximum number of results (default: 50).
:param offset: Return results starting from this index (default: 0).
:return: .. schema:: spotify.SpotifyAlbumSchema(many=True)
"""
return SpotifyAlbumSchema().dump(
self._spotify_paginate_results(
'/v1/me/albums',
limit=limit,
offset=offset,
),
many=True,
)
[docs] @action
def save_albums(self, resources: Iterable[str]):
"""
Save a list of albums to the user's collection.
:param resources: Spotify IDs or URIs of the albums to save.
"""
self.spotify_user_call(
'/v1/me/albums',
method='put',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def remove_albums(self, resources: Iterable[str]):
"""
Remove a list of albums from the user's collection.
:param resources: Spotify IDs or URIs of the albums to remove.
"""
self.spotify_user_call(
'/v1/me/albums',
method='delete',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def get_tracks(self, limit: int = 100, offset: int = 0) -> List[dict]:
"""
Get the list of tracks saved by the user.
:param limit: Maximum number of results (default: 100).
:param offset: Return results starting from this index (default: 0).
:return: .. schema:: spotify.SpotifyTrackSchema(many=True)
"""
return [
SpotifyTrackSchema().dump(item['track'])
for item in self._spotify_paginate_results(
'/v1/me/tracks', limit=limit, offset=offset
)
]
[docs] @action
def save_tracks(self, resources: Iterable[str]):
"""
Save a list of tracks to the user's collection.
:param resources: Spotify IDs or URIs of the tracks to save.
"""
self.spotify_user_call(
'/v1/me/tracks',
method='put',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def remove_tracks(self, resources: Iterable[str]):
"""
Remove a list of tracks from the user's collection.
:param resources: Spotify IDs or URIs of the tracks to remove.
"""
self.spotify_user_call(
'/v1/me/tracks',
method='delete',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def get_episodes(self, limit: int = 50, offset: int = 0) -> List[dict]:
"""
Get the list of episodes saved by the user.
:param limit: Maximum number of results (default: 50).
:param offset: Return results starting from this index (default: 0).
:return: .. schema:: spotify.SpotifyEpisodeSchema(many=True)
"""
return SpotifyEpisodeSchema().dump(
self._spotify_paginate_results(
'/v1/me/episodes',
limit=limit,
offset=offset,
),
many=True,
)
[docs] @action
def save_episodes(self, resources: Iterable[str]):
"""
Save a list of episodes to the user's collection.
:param resources: Spotify IDs or URIs of the episodes to save.
"""
self.spotify_user_call(
'/v1/me/episodes',
method='put',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def remove_episodes(self, resources: Iterable[str]):
"""
Remove a list of episodes from the user's collection.
:param resources: Spotify IDs or URIs of the episodes to remove.
"""
self.spotify_user_call(
'/v1/me/episodes',
method='delete',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def get_shows(self, limit: int = 50, offset: int = 0) -> List[dict]:
"""
Get the list of shows saved by the user.
:param limit: Maximum number of results (default: 50).
:param offset: Return results starting from this index (default: 0).
:return: .. schema:: spotify.SpotifyShowSchema(many=True)
"""
return SpotifyShowSchema().dump(
self._spotify_paginate_results(
'/v1/me/shows',
limit=limit,
offset=offset,
),
many=True,
)
[docs] @action
def save_shows(self, resources: Iterable[str]):
"""
Save a list of shows to the user's collection.
:param resources: Spotify IDs or URIs of the shows to save.
"""
self.spotify_user_call(
'/v1/me/shows',
method='put',
json={'ids': self._uris_to_id(*resources)},
)
[docs] @action
def remove_shows(self, resources: Iterable[str]):
"""
Remove a list of shows from the user's collection.
:param resources: Spotify IDs or URIs of the shows to remove.
"""
self.spotify_user_call(
'/v1/me/shows',
method='delete',
json={'ids': self._uris_to_id(*resources)},
)
