PK! spoffy/__init__.pyfrom spoffy.spotify import SyncSpotify, AsyncSpotify from spoffy.client.base import SyncClient, AsyncClient from spoffy.client.common import ClientCommon from spoffy.exceptions import ( SpotifyException, SpotifyUnauthorized, SpotifyPremiumRequired, ) __version__ = "0.1.0" __all__ = [ "SyncSpotify", "AsyncSpotify", "SyncClient", "AsyncClient", "ClientCommon", "SpotifyException", "SpotifyUnauthorized", "SpotifyPremiumRequired", ] PK!spoffy/client/__init__.pyPK!n&&spoffy/client/base.pyfrom abc import abstractmethod from spoffy.client.common import ClientCommon from spoffy.sansio import Response from spoffy.sansio import Request class AsyncClient(ClientCommon): """ Base class for async spotify client. Subclasses must implement the :meth:`~AsyncClient.request` method """ @abstractmethod async def request(self, request: Request) -> Response: pass class SyncClient(ClientCommon): """ Base class for a sync spotify client. Subclasses must implement the :meth:`~SyncClient.request` method """ @abstractmethod def request(self, request: Request) -> Response: pass SyncClient.__doc__ += ClientCommon.__doc__.split("----")[1] # type: ignore AsyncClient.__doc__ += ClientCommon.__doc__.split("----")[1] # type: ignore PK!0ӣjspoffy/client/common.pyfrom typing import Optional, Union, MutableMapping from cattr import Converter from spoffy.models import Token from spoffy.sansio import Response from spoffy.sansio import Request class ClientCommon: """ A sansio client implementation. This does no IO on its own. ---- :param access_token: A Spotify access token Supersedes the `token` parameter :param token: A token object, either for user or client :param client_id: Client ID for oauth login :param client_secret: Client secret for oauth login :param redirect_uri: Redirect URI for oauth login :param scope: Space separated list of scopes for oauth login :param state: State string for oauth login """ #: Base authorize url authorize_url = "https://accounts.spotify.com/authorize" #: Token url for oauth and refresh token_url = "https://accounts.spotify.com/api/token" #: URL prefix used for all Spotify API calls #: This can be changed if using a proxy base_url = "https://api.spotify.com/v1" def __init__( self, access_token: Optional[str] = None, token: Optional[Token] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None, scope: Optional[str] = None, state: Optional[str] = None, ) -> None: self._access_token = access_token self.token = token self.client_id = client_id self.client_secret = client_secret self.redirect_uri = redirect_uri self.scope = scope self.state = state self._converter = Converter() def prepare_request( self, method: str, url: str, params: Optional[MutableMapping] = None, body: Optional[Union[bytes, MutableMapping]] = None, headers: Optional[MutableMapping[str, str]] = None, access_token: Optional[str] = None, ) -> Request: """ Create a request object from the given arguments :param method: The request http method ( ``GET`` / ``POST`` / ``PUT`` / ``DELETE`` ) :param url: The request URL, relative URLs get prefixed with :py:attr:`~base_url` :param params: Dict of query parameters to add to URL :param body: Request body as either bytes or json serializable dict :param headers: Dict of headers to add to request :param access_token: Override client access token If not provided and no explicit ``Authorization`` header, the :py:attr:`~access_token` stored on this instance is used """ if url.startswith("http://") or url.startswith("https://"): pass else: url = self.base_url + url headers = headers or {} if not access_token and "Authorization" not in headers: access_token = self.access_token return Request( method=method, url=url, params=params, body=body, headers=headers, access_token=access_token, ) @property def access_token(self) -> Union[str, None]: """ Property to get the current access token Returns either the access_token passed to the constructor or the access token from this client's :py:attr:`~token` object. Returns ``None`` if no token """ if self._access_token: return self._access_token elif self.token: return self.token.access_token return None @access_token.setter def access_token(self, access_token: str): self.access_token = access_token def load(self, response: Response, target): """ Load a response data to a ``SpotifyObject`` object :param response: The response to load :param target: The target class """ if target is None: return None data = response.json if data is None: return None return target(**data) PK!spoffy/clientmodule.pyfrom spoffy.oldclient import Client class ClientModule: """ A base client module :param client: A `Client` instance """ def __init__(self, client: Client) -> None: self.client = client PK!ͱspoffy/exceptions.pyfrom typing import Optional as Opt, Dict class SpotifyException(Exception): """ Base exception for any Spotify API errors :param message: The error message from the API :param status_code: The HTTP status code of the error response :param url: The URL of the failed request :param request_method: The HTTP method of the failed request :param reason: The error reason as returned from the API (not included on all errors) :param error_description: A more detailed error description as returned from the API (not included on all errors) """ def __init__( self, message: str, *, request_method: str, request_url: str, status_code: int, headers: Dict[str, str], reason: Opt[str] = None, error_description: Opt[str] = None, ): self.message = message self.status_code = status_code self.headers = headers self.reason = reason self.error_description = error_description self.request_method = request_method self.request_url = request_url def __str__(self): return "{status_code}: {message} ({method} {url})".format( status_code=self.status_code, message=self.message, method=self.request_method, url=self.request_url, ) class SpotifyUnauthorized(SpotifyException): """ Raised on authorization errors such as when an invalid access token is used """ class SpotifyPremiumRequired(SpotifyException): """ Raise when attempting to call and endpoint which requires a premium subscription with a free account access token """ PK!spoffy/io/__init__.pyPK![ht t spoffy/io/aiohttp.pyfrom typing import Optional import aiohttp from spoffy.models import Token from spoffy.client.base import AsyncClient, ClientCommon from spoffy.sansio import Request, Response from spoffy.spotify import AsyncSpotify class AioHttpClient(AsyncClient): """ Client implementation using requests as a http backend :param session: A :class:`~aiohttp.ClientSession` object """ def __init__( self, *, session: aiohttp.ClientSession, access_token: Optional[str] = None, token: Optional[Token] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None, scope: Optional[str] = None, state: Optional[str] = None, ) -> None: self.session = session super().__init__( access_token, token, client_id, client_secret, redirect_uri, scope, state, ) async def request(self, request: Request) -> Response: async with self.session.request( method=request.method, url=request.url, data=request.body, headers=request.headers, ) as resp: response = Response( request, resp.status, resp.headers, await resp.read() ) response.raise_for_status() return response def make_spotify( *, session: aiohttp.ClientSession, access_token: Optional[str] = None, token: Optional[Token] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None, scope: Optional[str] = None, state: Optional[str] = None, ) -> AsyncSpotify: """ Convenience factory to build Spotify API wrapper using the requests http library. Accepts all arguments of :class:`~AioHttpClient` """ return AsyncSpotify( AioHttpClient( session=session, access_token=access_token, token=token, client_id=client_id, client_secret=client_secret, redirect_uri=redirect_uri, scope=scope, state=state, ) ) AioHttpClient.__doc__ += ClientCommon.__doc__.split("----")[1] # type: ignore make_spotify.__doc__ += ClientCommon.__doc__.split("----")[1] # type: ignore PK!rю spoffy/io/requests.pyfrom typing import Optional import requests from spoffy.models import Token from spoffy.client.base import SyncClient, ClientCommon from spoffy.sansio import Request, Response from spoffy.spotify import SyncSpotify class RequestsClient(SyncClient): """ Client implementation using requests as a http backend :param session: A :class:`~requests.Session` object """ def __init__( self, *, session: Optional[requests.Session] = None, access_token: Optional[str] = None, token: Optional[Token] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None, scope: Optional[str] = None, state: Optional[str] = None, ) -> None: self.session: requests.Session = session or requests # type: ignore super().__init__( access_token, token, client_id, client_secret, redirect_uri, scope, state, ) def request(self, request: Request) -> Response: resp = self.session.request( method=request.method, url=request.url, data=request.body, headers=request.headers, ) response = Response( request, resp.status_code, resp.headers, resp.content ) response.raise_for_status() return response def make_spotify( *, session: Optional[requests.Session] = None, access_token: Optional[str] = None, token: Optional[Token] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None, scope: Optional[str] = None, state: Optional[str] = None, ) -> SyncSpotify: """ Convenience factory to build Spotify API wrapper using the requests http library. Accepts all arguments of :class:`~RequestsClient` """ return SyncSpotify( RequestsClient( session=session, access_token=access_token, token=token, client_id=client_id, client_secret=client_secret, redirect_uri=redirect_uri, scope=scope, state=state, ) ) RequestsClient.__doc__ += ClientCommon.__doc__.split("----")[1] # type: ignore make_spotify.__doc__ += ClientCommon.__doc__.split("----")[1] # type: ignore PK!^rrspoffy/models/__init__.pyfrom spoffy.models.paging import Paging from spoffy.models.playlists import ( Playlist, PlaylistSimple, PlaylistTracksPaging, ) from spoffy.models.core import ( Artist, Track, Album, AlbumTracksPaging, ArtistAlbumsPaging, ) from spoffy.models.collections import TracksCollection, AudioFeaturesCollection from spoffy.models.audiofeatures import AudioFeatures from spoffy.models.search import SearchResults from spoffy.models.personalization import TopTracksPaging, TopArtistsPaging from spoffy.models.library import SavedAlbumsPaging, SavedTracksPaging from spoffy.models.player import ( Device, CurrentPlayback, DevicesCollection, PlayHistoryPaging, ) from spoffy.models.token import ClientCredentialsToken, UserToken, Token from spoffy.models.users import UserPublic, UserPrivate __all__ = [ "Playlist", "PlaylistTracksPaging", "PlaylistSimple", "Artist", "Track", "Album", "AlbumTracksPaging", "ArtistAlbumsPaging", "TracksCollection", "AudioFeaturesCollection", "AudioFeatures", "SearchResults", "TopTracksPaging", "TopArtistsPaging", "SavedAlbumsPaging", "SavedTracksPaging", "Device", "CurrentPlayback", "DevicesCollection", "PlayHistoryPaging", "UserToken", "ClientCredentialsToken", "UserPublic", "UserPrivate", "Token", "Paging", ] PK!-,8spoffy/models/audiofeatures.pyfrom spoffy.models.base import SpotifyObject class AudioFeatures(SpotifyObject): duration_ms: int key: int mode: int time_signature: int acousticness: float danceability: float energy: float instrumentalness: float liveness: float loudness: float speechiness: float valence: float tempo: float id: str uri: str track_href: str analysis_url: str type: str PK!Qwffspoffy/models/base.pyimport json from pprint import pformat from typing import Dict, Any, List from cattr import Converter class SpotifyObject: __spotify_object__ = True def __init__(self, **d: Dict[str, Any]) -> None: # type_hints = getattr(self.__class__, "__annotations__", {}) self.__annos = getattr(self.__class__, "__annotations__", {}) self.__attrs: List[str] = [] for key, value in d.items(): self.__attrs.append(key) if isinstance(value, dict): setattr(self, key, self.__convert_nested(key, value)) elif isinstance(value, list) and value: if isinstance(value[0], dict): setattr(self, key, self.__convert_nested_list(key, value)) else: setattr(self, key, value) else: setattr(self, key, value) def __convert_nested(self, key, value: dict): return converter.structure(value, self.__annos.get(key, SpotifyObject)) def __convert_nested_list(self, key, value: List[dict]): return converter.structure( value, self.__annos.get(key, List[SpotifyObject]) ) def to_dict(self) -> Dict[str, Any]: result = {} for key in self.__attrs: value = getattr(self, key) if hasattr(value, "to_dict"): value = value.to_dict() elif isinstance(value, list) and value: dict_list = hasattr(value[0], "to_dict") if dict_list: value = [item.to_dict() for item in value] result[key] = value return result def __str__(self): return json.dumps(self.to_dict(), indent=2) def __repr__(self): return "<{} ({})>".format( self.__class__.__name__, pformat(self.to_dict()) ) def convert_spotobject(obj, cl): return cl(**obj) def is_obj(cls): return getattr(cls, "__spotify_object__", False) converter = Converter() converter.register_structure_hook_func(is_obj, convert_spotobject) # converter.register_structure_hook(SpotifyObject, convert_spotobject) PK!^w.[[spoffy/models/collections.pyfrom typing import List, Union from spoffy.models.core import Track from spoffy.models.audiofeatures import AudioFeatures from spoffy.models.base import SpotifyObject class TracksCollection(SpotifyObject): tracks: List[Union[Track, None]] class AudioFeaturesCollection(SpotifyObject): audio_features: List[Union[AudioFeatures, None]] PK!\Hspoffy/models/compat.pyimport sys from typing import Sequence version_info = sys.version_info[0:3] is_py37 = version_info[:2] == (3, 7) if is_py37: from typing import List, Union, _GenericAlias # type: ignore def is_union(obj): return ( obj is Union or isinstance(obj, _GenericAlias) and obj.__origin__ is Union ) def is_list(type): return ( type is List or type.__class__ is _GenericAlias and issubclass(type.__origin__, List) ) def is_sequence(type): return ( type is List or type.__class__ is _GenericAlias and issubclass(type.__origin__, Sequence) ) else: from typing import _Union, List # type: ignore def is_union(obj): """Return true if the object is a union. """ return isinstance(obj, _Union) def is_list(type): return issubclass(type, List) def is_sequence(type): return issubclass(type, Sequence) PK!Z= = spoffy/models/core.pyfrom typing import List, Optional as Opt from spoffy.models.image import Image from spoffy.models.paging import OffsetPaging from spoffy.models.base import SpotifyObject class Copyright(SpotifyObject): text: str type: str class ExternalUrls(SpotifyObject): spotify: str class Restrictions(SpotifyObject): reason: str class ExternalIds(SpotifyObject): upc: Opt[str] = None class TrackExternalIds(SpotifyObject): isrc: str class Followers(SpotifyObject): href: Opt[str] total: int class ArtistSimple(SpotifyObject): id: str external_urls: ExternalUrls href: str name: str type: str uri: str class Artist(ArtistSimple): followers: Followers genres: List[str] images: List[Image] popularity: int class TrackLink(SpotifyObject): external_urls: ExternalUrls href: str id: str type: str uri: str class TrackSimple(SpotifyObject): id: str name: str artists: List[ArtistSimple] disc_number: int duration_ms: int explicit: bool external_urls: ExternalUrls href: str preview_url: Opt[str] track_number: int type: str is_local: bool uri: str available_markets: Opt[List[str]] = None linked_from: Opt[TrackLink] = None is_playable: Opt[bool] = None class AlbumBase(SpotifyObject): id: str name: str album_type: str artists: List[ArtistSimple] external_urls: ExternalUrls href: str images: List[Image] release_date: str release_date_precision: str type: str uri: str total_tracks: int class AlbumSimple(AlbumBase): restrictions: Opt[Restrictions] = None available_markets: Opt[List[str]] = None class Track(TrackSimple): explicit: bool album: AlbumSimple popularity: int external_ids: TrackExternalIds available_markets: Opt[List[str]] = None linked_from: Opt[TrackLink] = None is_playable: Opt[bool] = None class AlbumTracksPaging(OffsetPaging): items: List[TrackSimple] class Album(AlbumBase): copyrights: List[Copyright] external_ids: ExternalIds genres: List[str] label: str popularity: int tracks: AlbumTracksPaging restrictions: Opt[Restrictions] = None available_markets: Opt[List[str]] = None class ArtistAlbumsPaging(OffsetPaging): items: List[AlbumSimple] PK!B蕡spoffy/models/image.pyfrom typing import Optional as Opt from spoffy.models.base import SpotifyObject class Image(SpotifyObject): url: str width: Opt[int] = None height: Opt[int] = None PK!zspoffy/models/library.pyfrom typing import List from spoffy.models.core import Track, Album from spoffy.models.paging import OffsetPaging from spoffy.models.base import SpotifyObject class SavedTrack(SpotifyObject): added_at: str track: Track class SavedTracksPaging(OffsetPaging): items: List[SavedTrack] class SavedAlbum(SpotifyObject): added_at: str album: Album class SavedAlbumsPaging(OffsetPaging): items: List[SavedAlbum] PK!#spoffy/models/paging.pyfrom typing import Optional as Opt, List, Any from spoffy.models.base import SpotifyObject class Paging(SpotifyObject): href: str limit: int next: Opt[str] previous: Opt[str] items: List[Any] class OffsetPaging(Paging): """ Offset based paging object :param href: Url to this object :param href: URL to this object :param limit: Max number of items per page :param offset: The paging offset :param next: URL to next page (if available) :param previous: URL to previous page (if available) """ # Jedi doesn't support generics, do this for now href: str limit: int offset: int total: int next: Opt[str] previous: Opt[str] class Cursor(SpotifyObject): after: str before: str class CursorPaging(Paging): """ Cursor based paging object :param href: URL to this object :param limit: Max number of items per page :param cursors: Cursor object for pagination :param next: URL to next page (if available) :param previous: URL to previous page (if available) """ href: str limit: int cursors: Cursor next: Opt[str] = None previous: Opt[str] = None PK!Շ spoffy/models/personalization.pyfrom typing import List from spoffy.models.core import Track, Artist from spoffy.models.paging import OffsetPaging class TopArtistsPaging(OffsetPaging): items: List[Artist] class TopTracksPaging(OffsetPaging): items: List[Track] PK!hRsspoffy/models/player.pyfrom typing import Optional as Opt, List, Dict from spoffy.models.core import Track, ExternalUrls from spoffy.models.paging import CursorPaging from spoffy.models.base import SpotifyObject class Device(SpotifyObject): id: Opt[str] is_active: bool is_private_session: bool is_restricted: bool name: str type: str volume_percent: Opt[int] class DevicesCollection(SpotifyObject): devices: List[Device] class Context(SpotifyObject): uri: str href: Opt[str] external_urls: Opt[ExternalUrls] type: str class CurrentPlayback(SpotifyObject): device: Device repeat_state: str shuffle_state: bool context: Opt[Context] timestamp: int progress_ms: Opt[int] is_playing: bool item: Opt[Track] currently_playing_type: str actions: Dict # Undocumented param, keep as dict class PlayHistoryItem(SpotifyObject): track: Track played_at: str context: Opt[Context] class PlayHistoryPaging(CursorPaging): items: List[PlayHistoryItem] PK!>Zˋ{{spoffy/models/playlists.pyfrom typing import List, Optional as Opt from spoffy.models.image import Image from spoffy.models.users import UserBase from spoffy.models.core import Track, TrackLink, ExternalUrls, Followers from spoffy.models.paging import OffsetPaging from spoffy.models.base import SpotifyObject class PlaylistOwner(UserBase): display_name: str class PlaylistTrackAddedBy(UserBase): pass class VideoThumbnail(SpotifyObject): url: Opt[str] class PlaylistTrackTrack(Track): track: bool episode: bool available_markets: Opt[List[str]] = None linked_from: Opt[TrackLink] = None is_playable: Opt[bool] = None class PlaylistTrack(SpotifyObject): added_at: Opt[str] # todo: datetime added_by: Opt[PlaylistTrackAddedBy] is_local: bool primary_color: Opt[str] = None track: PlaylistTrackTrack video_thumbnail: VideoThumbnail class PlaylistTracksPaging(OffsetPaging): items: List[PlaylistTrack] class PlaylistBase(SpotifyObject): """ :param collaborative: :param external_urls: :param href: :param id: :param images: :param name: :param owner: :param public: :param snapshot_id: :param type: :param uri: """ collaborative: bool external_urls: ExternalUrls href: str id: str images: List[Image] name: str owner: PlaylistOwner public: bool snapshot_id: str type: str uri: str class TracksHref(SpotifyObject): href: str total: int class PlaylistSimple(PlaylistBase): """ A simplified playlist object :param tracks: """ tracks: TracksHref PlaylistSimple.__doc__ += PlaylistBase.__doc__ # type: ignore class Playlist(PlaylistBase): """ A complete playlist object :param description: :param followers: :param tracks: """ description: Opt[str] followers: Followers tracks: PlaylistTracksPaging PK!+9spoffy/models/search.pyfrom typing import List from spoffy.models.paging import OffsetPaging from spoffy.models.core import Artist, Track, AlbumSimple from spoffy.models.playlists import PlaylistSimple from spoffy.models.base import SpotifyObject class SearchArtistsPaging(OffsetPaging): items: List[Artist] class SearchTracksPaging(OffsetPaging): items: List[Track] class SearchAlbumsPaging(OffsetPaging): items: List[AlbumSimple] class SearchPlaylistsPaging(OffsetPaging): items: List[PlaylistSimple] class SearchResults(SpotifyObject): artists: SearchArtistsPaging tracks: SearchTracksPaging albums: SearchAlbumsPaging playlists: SearchPlaylistsPaging PK!#bbspoffy/models/token.pyimport time from typing import Optional as Opt from spoffy.models.base import SpotifyObject class Token(SpotifyObject): access_token: str token_type: str scope: str expires_in: int expires_at: int refresh_token: Opt[str] = None def __init__(self, **d): super().__init__(**d) if not hasattr(self, "expires_at"): self.expires_at = int(time.time()) + self.expires_in - 1 class ClientCredentialsToken(Token): """ A token generated through client credentials :param access_token: A Spotify access token :param token_type: The token type (always `Bearer`) :param scope: Scopes of this token (space seperated list) :param expires_in: Expiry time in seconds from token generation :param expires_at: Unix timestamp in seconds when this token expires """ pass class UserToken(Token): """ A token generated for user through OAuth login flow :param access_token: A Spotify access token :param token_type: The token type (always `Bearer`) :param scope: Scopes of this token (space seperated list) :param expires_in: Expiry time in seconds from token generation :param expires_at: Unix timestamp in seconds when this token expires :param refresh_token: A Spotify refresh token """ refresh_token: str PK! **spoffy/models/users.pyfrom typing import List, Optional as Opt from spoffy.models.core import ExternalUrls, Followers from spoffy.models.image import Image from spoffy.models.base import SpotifyObject class UserBase(SpotifyObject): external_urls: ExternalUrls href: str id: str type: str uri: str class UserPublic(UserBase): followers: Followers images: List[Image] display_name: Opt[str] class UserPrivate(UserPublic): birthdate: Opt[str] = None country: Opt[str] = None email: Opt[str] = None product: Opt[str] = None PK!spoffy/modules/__init__.pyPK!P]mmspoffy/modules/apimodule.pyfrom abc import abstractmethod from typing import TypeVar, Awaitable, Any from spoffy.sansio import Request from spoffy.client.base import AsyncClient, SyncClient TLoaded = TypeVar("TLoaded") class AsyncApiModule: def __init__(self, client: AsyncClient): self.client = client self.b = self.__builder_class__(client) async def _make_request(self, req: Request, target): return self.client.load(await self.client.request(req), target) async def _assign_result(self, attribute: str, response: Awaitable): setattr(self.client, attribute, await response) @property @abstractmethod def __builder_class__(self): pass class ApiModule: def __init__(self, client: SyncClient): self.client = client self.b = self.__builder_class__(client) def _make_request(self, req: Request, target): return self.client.load(self.client.request(req), target) def _assign_result(self, attribute: str, response: Any): setattr(self.client, attribute, response) @property @abstractmethod def __builder_class__(self): pass PK!URHHspoffy/modules/mixins.pyfrom urlobject import URLObject class AuthMixin: def get_authorize_url(self, **kwargs) -> str: params = dict( client_id=self.client.client_id, # type: ignore redirect_uri=self.client.redirect_uri, # type: ignore scope=self.client.scope, # type: ignore state=self.client.state, # type: ignore response_type="code", ) params.update(kwargs) return str( URLObject( self.client.authorize_url # type: ignore ).add_query_params(**params) ) PK!p|p|spoffy/modules/modules.pyfrom typing import Sequence, Optional, Dict, Union from spoffy import models from spoffy.modules.apimodule import ApiModule, AsyncApiModule from spoffy.modules import sansio as builders, mixins class Artists(ApiModule): __builder_class__ = builders.Artists def artist(self, artist_id: str) -> models.Artist: """ Get an artist by Spotify ID :param artist_id: """ return self._make_request( self.b.artist(artist_id=artist_id), models.Artist ) def artist_albums( self, artist_id: str, include_groups: Sequence[str] = None, market: str = None, ) -> models.ArtistAlbumsPaging: """ Get all artist's albums :param artist_id: :param include_groups: :param market: """ return self._make_request( self.b.artist_albums( artist_id=artist_id, include_groups=include_groups, market=market, ), models.ArtistAlbumsPaging, ) class AsyncArtists(AsyncApiModule): __builder_class__ = builders.Artists async def artist(self, artist_id: str) -> models.Artist: """ Get an artist by Spotify ID :param artist_id: """ return await self._make_request( self.b.artist(artist_id=artist_id), models.Artist ) async def artist_albums( self, artist_id: str, include_groups: Sequence[str] = None, market: str = None, ) -> models.ArtistAlbumsPaging: """ Get all artist's albums :param artist_id: :param include_groups: :param market: """ return await self._make_request( self.b.artist_albums( artist_id=artist_id, include_groups=include_groups, market=market, ), models.ArtistAlbumsPaging, ) class Playlists(ApiModule): __builder_class__ = builders.Playlists def playlist( self, playlist_id: str, market: Optional[str] = None, fields: Optional[str] = None, ) -> models.Playlist: return self._make_request( self.b.playlist( playlist_id=playlist_id, market=market, fields=fields ), models.Playlist, ) def playlist_tracks( self, playlist_id: str, market: Optional[str] = None, fields: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, ) -> models.PlaylistTracksPaging: return self._make_request( self.b.playlist_tracks( playlist_id=playlist_id, market=market, fields=fields, limit=limit, offset=offset, ), models.PlaylistTracksPaging, ) class AsyncPlaylists(AsyncApiModule): __builder_class__ = builders.Playlists async def playlist( self, playlist_id: str, market: Optional[str] = None, fields: Optional[str] = None, ) -> models.Playlist: return await self._make_request( self.b.playlist( playlist_id=playlist_id, market=market, fields=fields ), models.Playlist, ) async def playlist_tracks( self, playlist_id: str, market: Optional[str] = None, fields: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, ) -> models.PlaylistTracksPaging: return await self._make_request( self.b.playlist_tracks( playlist_id=playlist_id, market=market, fields=fields, limit=limit, offset=offset, ), models.PlaylistTracksPaging, ) class Albums(ApiModule): __builder_class__ = builders.Albums def get_album( self, album_id: str, market: Optional[str] = None ) -> models.Album: """ Get a single album :param album_id: The Spotify album ID to fetch :param market: ISO2A country to enable relinking """ return self._make_request( self.b.get_album(album_id=album_id, market=market), models.Album ) def get_album_tracks( self, album_id: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, ) -> models.AlbumTracksPaging: """ Get tracks from the given album :param album_id: The Spotify album ID to fetch tracks from :param market: ISO2A country to enable relinking :param limit: Max number of results to retrieve :param offset: Pagination offset """ return self._make_request( self.b.get_album_tracks( album_id=album_id, market=market, limit=limit, offset=offset ), models.AlbumTracksPaging, ) class AsyncAlbums(AsyncApiModule): __builder_class__ = builders.Albums async def get_album( self, album_id: str, market: Optional[str] = None ) -> models.Album: """ Get a single album :param album_id: The Spotify album ID to fetch :param market: ISO2A country to enable relinking """ return await self._make_request( self.b.get_album(album_id=album_id, market=market), models.Album ) async def get_album_tracks( self, album_id: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, ) -> models.AlbumTracksPaging: """ Get tracks from the given album :param album_id: The Spotify album ID to fetch tracks from :param market: ISO2A country to enable relinking :param limit: Max number of results to retrieve :param offset: Pagination offset """ return await self._make_request( self.b.get_album_tracks( album_id=album_id, market=market, limit=limit, offset=offset ), models.AlbumTracksPaging, ) class Tracks(ApiModule): __builder_class__ = builders.Tracks def get_track( self, track_id: str, market: Optional[str] = None ) -> models.Track: """ Get a single track :param track_id: The Spotify track ID to fetch :param market: ISO2A country to enable track relinking """ return self._make_request( self.b.get_track(track_id=track_id, market=market), models.Track ) def get_many_tracks( self, track_ids: Sequence[str], market: Optional[str] = None ) -> models.TracksCollection: """ Get several tracks by ID :param track_ids: List of up to 50 track IDs :param market: ISO2A country to enable track relinking """ return self._make_request( self.b.get_many_tracks(track_ids=track_ids, market=market), models.TracksCollection, ) def get_audio_features(self, track_id: str) -> models.AudioFeatures: """ Get audio features for a single track :param track_id: The Spotify track ID to fetch audio features for """ return self._make_request( self.b.get_audio_features(track_id=track_id), models.AudioFeatures ) def get_many_audio_features( self, track_ids: Sequence[str] ) -> models.AudioFeaturesCollection: """ Get audio features for several tracks :param track_ids: List of up to 50 track IDs """ return self._make_request( self.b.get_many_audio_features(track_ids=track_ids), models.AudioFeaturesCollection, ) class AsyncTracks(AsyncApiModule): __builder_class__ = builders.Tracks async def get_track( self, track_id: str, market: Optional[str] = None ) -> models.Track: """ Get a single track :param track_id: The Spotify track ID to fetch :param market: ISO2A country to enable track relinking """ return await self._make_request( self.b.get_track(track_id=track_id, market=market), models.Track ) async def get_many_tracks( self, track_ids: Sequence[str], market: Optional[str] = None ) -> models.TracksCollection: """ Get several tracks by ID :param track_ids: List of up to 50 track IDs :param market: ISO2A country to enable track relinking """ return await self._make_request( self.b.get_many_tracks(track_ids=track_ids, market=market), models.TracksCollection, ) async def get_audio_features(self, track_id: str) -> models.AudioFeatures: """ Get audio features for a single track :param track_id: The Spotify track ID to fetch audio features for """ return await self._make_request( self.b.get_audio_features(track_id=track_id), models.AudioFeatures ) async def get_many_audio_features( self, track_ids: Sequence[str] ) -> models.AudioFeaturesCollection: """ Get audio features for several tracks :param track_ids: List of up to 50 track IDs """ return await self._make_request( self.b.get_many_audio_features(track_ids=track_ids), models.AudioFeaturesCollection, ) class Users(ApiModule): __builder_class__ = builders.Users def user(self, user_id: str) -> models.UserPublic: return self._make_request( self.b.user(user_id=user_id), models.UserPublic ) def me(self) -> models.UserPrivate: return self._make_request(self.b.me(), models.UserPrivate) class AsyncUsers(AsyncApiModule): __builder_class__ = builders.Users async def user(self, user_id: str) -> models.UserPublic: return await self._make_request( self.b.user(user_id=user_id), models.UserPublic ) async def me(self) -> models.UserPrivate: return await self._make_request(self.b.me(), models.UserPrivate) class Player(ApiModule): __builder_class__ = builders.Player def play( self, device_id: Optional[str] = None, context_uri: Optional[str] = None, uris: Optional[Sequence[str]] = None, offset: Optional[Dict[str, Union[str, int]]] = None, position_ms: Optional[int] = None, ) -> None: """ Play an album, track or playlist on Spotify """ return self._make_request( self.b.play( device_id=device_id, context_uri=context_uri, uris=uris, offset=offset, position_ms=position_ms, ), None, ) def current_playback( self, market: Optional[str] = None ) -> models.CurrentPlayback: """ Get current user's current playback status """ return self._make_request( self.b.current_playback(market=market), models.CurrentPlayback ) def next_track(self, device_id: Optional[str] = None) -> None: """ Skip to next track in Spotify player """ return self._make_request(self.b.next_track(device_id=device_id), None) def previous_track(self, device_id: Optional[str] = None) -> None: """ Skip to previous track in Spotify player """ return self._make_request( self.b.previous_track(device_id=device_id), None ) def pause(self, device_id: Optional[str] = None) -> None: return self._make_request(self.b.pause(device_id=device_id), None) def set_volume( self, volume_percent: int, device_id: Optional[str] = None ) -> None: return self._make_request( self.b.set_volume( volume_percent=volume_percent, device_id=device_id ), None, ) def recently_played( self, limit: int = 20, before: Optional[int] = None, after: Optional[int] = None, ) -> models.PlayHistoryPaging: return self._make_request( self.b.recently_played(limit=limit, before=before, after=after), models.PlayHistoryPaging, ) class AsyncPlayer(AsyncApiModule): __builder_class__ = builders.Player async def play( self, device_id: Optional[str] = None, context_uri: Optional[str] = None, uris: Optional[Sequence[str]] = None, offset: Optional[Dict[str, Union[str, int]]] = None, position_ms: Optional[int] = None, ) -> None: """ Play an album, track or playlist on Spotify """ return await self._make_request( self.b.play( device_id=device_id, context_uri=context_uri, uris=uris, offset=offset, position_ms=position_ms, ), None, ) async def current_playback( self, market: Optional[str] = None ) -> models.CurrentPlayback: """ Get current user's current playback status """ return await self._make_request( self.b.current_playback(market=market), models.CurrentPlayback ) async def next_track(self, device_id: Optional[str] = None) -> None: """ Skip to next track in Spotify player """ return await self._make_request( self.b.next_track(device_id=device_id), None ) async def previous_track(self, device_id: Optional[str] = None) -> None: """ Skip to previous track in Spotify player """ return await self._make_request( self.b.previous_track(device_id=device_id), None ) async def pause(self, device_id: Optional[str] = None) -> None: return await self._make_request( self.b.pause(device_id=device_id), None ) async def set_volume( self, volume_percent: int, device_id: Optional[str] = None ) -> None: return await self._make_request( self.b.set_volume( volume_percent=volume_percent, device_id=device_id ), None, ) async def recently_played( self, limit: int = 20, before: Optional[int] = None, after: Optional[int] = None, ) -> models.PlayHistoryPaging: return await self._make_request( self.b.recently_played(limit=limit, before=before, after=after), models.PlayHistoryPaging, ) class Search(ApiModule): __builder_class__ = builders.Search def search( self, q: str, type: Sequence[str], market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: return self._make_request( self.b.search( q=q, type=type, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) def artists( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for artists. Equivalent to `search(q, types='artist', ...)` :param q: The search query """ return self._make_request( self.b.artists( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) def albums( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for albums. Equivalent to `search(q, types='album', ...)` :param q: The search query """ return self._make_request( self.b.albums( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) def tracks( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for tracks. Equivalent to `search(q, types='track', ...)` :param q: The search query """ return self._make_request( self.b.tracks( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) def playlists( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for playlists. Equivalent to `search(q, types='playlist', ...)` :param q: The search query """ return self._make_request( self.b.playlists( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) class AsyncSearch(AsyncApiModule): __builder_class__ = builders.Search async def search( self, q: str, type: Sequence[str], market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: return await self._make_request( self.b.search( q=q, type=type, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) async def artists( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for artists. Equivalent to `search(q, types='artist', ...)` :param q: The search query """ return await self._make_request( self.b.artists( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) async def albums( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for albums. Equivalent to `search(q, types='album', ...)` :param q: The search query """ return await self._make_request( self.b.albums( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) async def tracks( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for tracks. Equivalent to `search(q, types='track', ...)` :param q: The search query """ return await self._make_request( self.b.tracks( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) async def playlists( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> models.SearchResults: """ Convenience method to search for playlists. Equivalent to `search(q, types='playlist', ...)` :param q: The search query """ return await self._make_request( self.b.playlists( q=q, market=market, limit=limit, offset=offset, include_external=include_external, ), models.SearchResults, ) class Library(ApiModule): __builder_class__ = builders.Library def top_artists( self, limit: Optional[int] = 20, offset: Optional[int] = None, time_range: Optional[str] = None, ) -> models.TopArtistsPaging: """ Get current user's top artists :param limit: Max number of results :param offset: Pagination offset :time_range: Top artists over specified time period """ return self._make_request( self.b.top_artists( limit=limit, offset=offset, time_range=time_range ), models.TopArtistsPaging, ) def top_tracks( self, limit: Optional[int] = 20, offset: Optional[int] = None, time_range: Optional[str] = None, ) -> models.TopTracksPaging: """ Get current user's top artists :param limit: Max number of results :param offset: Pagination offset :time_range: Top artists over specified time period """ return self._make_request( self.b.top_tracks( limit=limit, offset=offset, time_range=time_range ), models.TopTracksPaging, ) def saved_albums( self, limit: Optional[int] = None, offset: Optional[int] = None, market: Optional[str] = None, ) -> models.SavedAlbumsPaging: return self._make_request( self.b.saved_albums(limit=limit, offset=offset, market=market), models.SavedAlbumsPaging, ) def saved_tracks( self, limit: Optional[int] = None, offset: Optional[int] = None, market: Optional[str] = None, ) -> models.SavedTracksPaging: return self._make_request( self.b.saved_tracks(limit=limit, offset=offset, market=market), models.SavedTracksPaging, ) def remove_saved_albums(self, ids: Sequence[str]) -> None: return self._make_request(self.b.remove_saved_albums(ids=ids), None) def add_saved_albums(self, ids: Sequence[str]) -> None: return self._make_request(self.b.add_saved_albums(ids=ids), None) def remove_saved_tracks(self, ids: Sequence[str]) -> None: return self._make_request(self.b.remove_saved_tracks(ids=ids), None) def add_saved_tracks(self, ids: Sequence[str]) -> None: return self._make_request(self.b.add_saved_tracks(ids=ids), None) class AsyncLibrary(AsyncApiModule): __builder_class__ = builders.Library async def top_artists( self, limit: Optional[int] = 20, offset: Optional[int] = None, time_range: Optional[str] = None, ) -> models.TopArtistsPaging: """ Get current user's top artists :param limit: Max number of results :param offset: Pagination offset :time_range: Top artists over specified time period """ return await self._make_request( self.b.top_artists( limit=limit, offset=offset, time_range=time_range ), models.TopArtistsPaging, ) async def top_tracks( self, limit: Optional[int] = 20, offset: Optional[int] = None, time_range: Optional[str] = None, ) -> models.TopTracksPaging: """ Get current user's top artists :param limit: Max number of results :param offset: Pagination offset :time_range: Top artists over specified time period """ return await self._make_request( self.b.top_tracks( limit=limit, offset=offset, time_range=time_range ), models.TopTracksPaging, ) async def saved_albums( self, limit: Optional[int] = None, offset: Optional[int] = None, market: Optional[str] = None, ) -> models.SavedAlbumsPaging: return await self._make_request( self.b.saved_albums(limit=limit, offset=offset, market=market), models.SavedAlbumsPaging, ) async def saved_tracks( self, limit: Optional[int] = None, offset: Optional[int] = None, market: Optional[str] = None, ) -> models.SavedTracksPaging: return await self._make_request( self.b.saved_tracks(limit=limit, offset=offset, market=market), models.SavedTracksPaging, ) async def remove_saved_albums(self, ids: Sequence[str]) -> None: return await self._make_request( self.b.remove_saved_albums(ids=ids), None ) async def add_saved_albums(self, ids: Sequence[str]) -> None: return await self._make_request(self.b.add_saved_albums(ids=ids), None) async def remove_saved_tracks(self, ids: Sequence[str]) -> None: return await self._make_request( self.b.remove_saved_tracks(ids=ids), None ) async def add_saved_tracks(self, ids: Sequence[str]) -> None: return await self._make_request(self.b.add_saved_tracks(ids=ids), None) class Auth(ApiModule, mixins.AuthMixin): __builder_class__ = builders.Auth def get_token_from_client_credentials( self ) -> models.ClientCredentialsToken: """ Get an authorization token from client credentials using :py:attr:`~client.client_id` and :py:attr:`~client.client_secret` """ return self._make_request( self.b.get_token_from_client_credentials(), models.ClientCredentialsToken, ) def get_token_from_refresh_token( self, refresh_token: str = None ) -> models.UserToken: """ Get a fresh authorization token from a refresh token :param refresh_token: A Spotify refresh token, if `None` will use the refresh token stored on client instance. """ return self._make_request( self.b.get_token_from_refresh_token(refresh_token=refresh_token), models.UserToken, ) def get_token_from_code( self, response_code: str, **kwargs ) -> models.UserToken: """ Get an authorization token from an OAuth response code :param response_code: The response code from redirect URL :param \\**kwargs: Additional keyword arguments that override instance attributes and are sent to the token API as query params """ return self._make_request( self.b.get_token_from_code(response_code=response_code, **kwargs), models.UserToken, ) def authorize_user(self, response_code: str, **kwargs): """ Authorize this API instance using a response code from oauth login """ self._assign_result( "token", self.get_token_from_code(response_code=response_code, **kwargs), ) def authorize_client(self): """ Authorize this API instance using its client ID and client Secret """ self._assign_result("token", self.get_token_from_client_credentials()) def refresh_authorization(self, refresh_token: str = None): """ :param refresh_token: Optional refresh token to use instead of the token stored on this instance """ self._assign_result( "token", self.get_token_from_refresh_token(refresh_token=refresh_token), ) class AsyncAuth(AsyncApiModule, mixins.AuthMixin): __builder_class__ = builders.Auth async def get_token_from_client_credentials( self ) -> models.ClientCredentialsToken: """ Get an authorization token from client credentials using :py:attr:`~client.client_id` and :py:attr:`~client.client_secret` """ return await self._make_request( self.b.get_token_from_client_credentials(), models.ClientCredentialsToken, ) async def get_token_from_refresh_token( self, refresh_token: str = None ) -> models.UserToken: """ Get a fresh authorization token from a refresh token :param refresh_token: A Spotify refresh token, if `None` will use the refresh token stored on client instance. """ return await self._make_request( self.b.get_token_from_refresh_token(refresh_token=refresh_token), models.UserToken, ) async def get_token_from_code( self, response_code: str, **kwargs ) -> models.UserToken: """ Get an authorization token from an OAuth response code :param response_code: The response code from redirect URL :param \\**kwargs: Additional keyword arguments that override instance attributes and are sent to the token API as query params """ return await self._make_request( self.b.get_token_from_code(response_code=response_code, **kwargs), models.UserToken, ) async def authorize_user(self, response_code: str, **kwargs): """ Authorize this API instance using a response code from oauth login """ await self._assign_result( "token", self.get_token_from_code(response_code=response_code, **kwargs), ) async def authorize_client(self): """ Authorize this API instance using its client ID and client Secret """ await self._assign_result( "token", self.get_token_from_client_credentials() ) async def refresh_authorization(self, refresh_token: str = None): """ :param refresh_token: Optional refresh token to use instead of the token stored on this instance """ await self._assign_result( "token", self.get_token_from_refresh_token(refresh_token=refresh_token), ) PK!¤DDspoffy/modules/sansio.pyfrom base64 import b64encode from typing import Sequence, Optional, Dict, Union from spoffy.sansio import Request from spoffy.client.common import ClientCommon from spoffy import models def returns(type_): def decorator(func): func.__spotify_object__ = type_ return func return decorator class RequestBuilder: def __init__(self, client: ClientCommon): self.client = client def b(self, *args, **kwargs): return self.client.prepare_request(*args, **kwargs) class Artists(RequestBuilder): @returns(models.Artist) def artist(self, artist_id: str) -> Request: """ Get an artist by Spotify ID :param artist_id: """ return self.b(method="GET", url="/artists/{}".format(artist_id)) @returns(models.ArtistAlbumsPaging) def artist_albums( self, artist_id: str, include_groups: Sequence[str] = None, market: str = None, ) -> Request: """ Get all artist's albums :param artist_id: :param include_groups: :param market: """ params = {} if include_groups: params["include_groups"] = ",".join(include_groups) if market is not None: params["market"] = market return self.b( method="GET", url="/artists/{}/albums".format(artist_id), params=params, ) class Playlists(RequestBuilder): @returns(models.Playlist) def playlist( self, playlist_id: str, market: Optional[str] = None, fields: Optional[str] = None, ) -> Request: return self.b( "GET", "/playlists/{}".format(playlist_id), params=_clear_nones(dict(market=market, fields=fields)), ) @returns(models.PlaylistTracksPaging) def playlist_tracks( self, playlist_id: str, market: Optional[str] = None, fields: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, ) -> Request: return self.b( "GET", "/playlists/{}/tracks".format(playlist_id), params=_clear_nones( dict(market=market, fields=fields, limit=limit, offset=offset) ), ) class Albums(RequestBuilder): """ Interface to the Spotify albums API https://developer.spotify.com/documentation/web-api/reference/albums/ """ @returns(models.Album) def get_album( self, album_id: str, market: Optional[str] = None ) -> Request: """ Get a single album :param album_id: The Spotify album ID to fetch :param market: ISO2A country to enable relinking """ params = dict(market=market) if market is not None else None return self.b("GET", "/albums/{}".format(album_id), params=params) @returns(models.AlbumTracksPaging) def get_album_tracks( self, album_id: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, ) -> Request: """ Get tracks from the given album :param album_id: The Spotify album ID to fetch tracks from :param market: ISO2A country to enable relinking :param limit: Max number of results to retrieve :param offset: Pagination offset """ params = _clear_nones(dict(market=market, limit=limit, offset=offset)) return self.b( "GET", "/albums/{}/tracks".format(album_id), params=params ) class Tracks(RequestBuilder): """ Interface to Spotify tracks API https://developer.spotify.com/documentation/web-api/reference/tracks """ @returns(models.Track) def get_track( self, track_id: str, market: Optional[str] = None ) -> Request: """ Get a single track :param track_id: The Spotify track ID to fetch :param market: ISO2A country to enable track relinking """ params = {"market": market} if market is not None else None return self.b("GET", "/tracks/{}".format(track_id), params=params) @returns(models.TracksCollection) def get_many_tracks( self, track_ids: Sequence[str], market: Optional[str] = None ) -> Request: """ Get several tracks by ID :param track_ids: List of up to 50 track IDs :param market: ISO2A country to enable track relinking """ return self.b( "GET", "/tracks", params=_clear_nones(dict(ids=",".join(track_ids), market=market)), ) @returns(models.AudioFeatures) def get_audio_features(self, track_id: str) -> Request: """ Get audio features for a single track :param track_id: The Spotify track ID to fetch audio features for """ return self.b("GET", "/audio-features/{}".format(track_id)) @returns(models.AudioFeaturesCollection) def get_many_audio_features(self, track_ids: Sequence[str]) -> Request: """ Get audio features for several tracks :param track_ids: List of up to 50 track IDs """ return self.b( "GET", "/audio-features", params={"ids": ",".join(track_ids)} ) class Users(RequestBuilder): @returns(models.UserPublic) def user(self, user_id: str) -> Request: return self.b("GET", "/users/{}".format(user_id)) @returns(models.UserPrivate) def me(self) -> Request: return self.b("GET", "/me") class Player(RequestBuilder): @returns(None) def play( self, device_id: Optional[str] = None, context_uri: Optional[str] = None, uris: Optional[Sequence[str]] = None, offset: Optional[Dict[str, Union[str, int]]] = None, position_ms: Optional[int] = None, ) -> Request: """ Play an album, track or playlist on Spotify """ body = _clear_nones( dict( context_uri=context_uri, uris=uris, offset=offset, position_ms=position_ms, ) ) params = dict(device_id=device_id) if device_id is not None else None return self.b("PUT", "/me/player/play", body=body, params=params) @returns(models.CurrentPlayback) # TODO: This returns None sometimes def current_playback(self, market: Optional[str] = None) -> Request: """ Get current user's current playback status """ return self.b( "GET", "/me/player", params=_clear_nones(dict(market=market)) ) @returns(None) def next_track(self, device_id: Optional[str] = None) -> Request: """ Skip to next track in Spotify player """ params = dict(device_id=device_id) if device_id is not None else None return self.b("PUT", "/me/player/next", params=params) @returns(None) def previous_track(self, device_id: Optional[str] = None) -> Request: """ Skip to previous track in Spotify player """ params = dict(device_id=device_id) if device_id is not None else None return self.b("PUT", "/me/player/previous", params=params) @returns(None) def pause(self, device_id: Optional[str] = None) -> Request: params = dict(device_id=device_id) if device_id is not None else None return self.b("PUT", "/me/player/pause", params=params) @returns(None) def set_volume( self, volume_percent: int, device_id: Optional[str] = None ) -> Request: params = _clear_nones( dict(device_id=device_id, volume_percent=volume_percent) ) return self.b("PUT", "/me/player/volume", params=params) @returns(models.PlayHistoryPaging) def recently_played( self, limit: int = 20, before: Optional[int] = None, after: Optional[int] = None, ) -> Request: params = _clear_nones(dict(limit=limit, before=before, after=after)) return self.b("GET", "/me/player/recently-played", params=params) class Search(RequestBuilder): """ Interface to the Spotify search API https://developer.spotify.com/documentation/web-api/reference/search/search/ """ @returns(models.SearchResults) def search( self, q: str, type: Sequence[str], market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> Request: params = _clear_nones( dict( q=q, type=",".join(type), market=market, limit=limit, offset=offset, include_external=include_external, ) ) return self.b("GET", "/search", params=params) @returns(models.SearchResults) def artists( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> Request: """ Convenience method to search for artists. Equivalent to `search(q, types='artist', ...)` :param q: The search query """ return self.search( q=q, type=["artist"], market=market, limit=limit, offset=offset, include_external=include_external, ) @returns(models.SearchResults) def albums( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> Request: """ Convenience method to search for albums. Equivalent to `search(q, types='album', ...)` :param q: The search query """ return self.search( q=q, type=["album"], market=market, limit=limit, offset=offset, include_external=include_external, ) @returns(models.SearchResults) def tracks( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> Request: """ Convenience method to search for tracks. Equivalent to `search(q, types='track', ...)` :param q: The search query """ return self.search( q=q, type=["track"], market=market, limit=limit, offset=offset, include_external=include_external, ) @returns(models.SearchResults) def playlists( self, q: str, market: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, include_external: Optional[str] = None, ) -> Request: """ Convenience method to search for playlists. Equivalent to `search(q, types='playlist', ...)` :param q: The search query """ return self.search( q=q, type=["playlist"], market=market, limit=limit, offset=offset, include_external=include_external, ) class Library(RequestBuilder): """ Manage user's saved tracks and albums https://developer.spotify.com/documentation/web-api/reference/library/ """ @returns(models.TopArtistsPaging) def top_artists( self, limit: Optional[int] = 20, offset: Optional[int] = None, time_range: Optional[str] = None, ) -> Request: """ Get current user's top artists :param limit: Max number of results :param offset: Pagination offset :time_range: Top artists over specified time period """ return self.b( "GET", "/me/top/artists", params=_clear_nones( dict(limit=limit, offset=offset, time_range=time_range) ), ) @returns(models.TopTracksPaging) def top_tracks( self, limit: Optional[int] = 20, offset: Optional[int] = None, time_range: Optional[str] = None, ) -> Request: """ Get current user's top artists :param limit: Max number of results :param offset: Pagination offset :time_range: Top artists over specified time period """ return self.b( "GET", "/me/top/tracks", params=_clear_nones( dict(limit=limit, offset=offset, time_range=time_range) ), ) @returns(models.SavedAlbumsPaging) def saved_albums( self, limit: Optional[int] = None, offset: Optional[int] = None, market: Optional[str] = None, ) -> Request: return self.b( "GET", "/me/albums", params=_clear_nones( dict(market=market, limit=limit, offset=offset) ), ) @returns(models.SavedTracksPaging) def saved_tracks( self, limit: Optional[int] = None, offset: Optional[int] = None, market: Optional[str] = None, ) -> Request: return self.b( "GET", "/me/tracks", params=_clear_nones( dict(limit=limit, offset=offset, market=market) ), ) @returns(None) def remove_saved_albums(self, ids: Sequence[str]) -> Request: return self.b("DELETE", "/me/albums", params=dict(ids=",".join(ids))) @returns(None) def add_saved_albums(self, ids: Sequence[str]) -> Request: return self.b("PUT", "/me/albums", params=dict(ids=",".join(ids))) @returns(None) def remove_saved_tracks(self, ids: Sequence[str]) -> Request: return self.b("DELETE", "/me/tracks", params=dict(ids=",".join(ids))) @returns(None) def add_saved_tracks(self, ids: Sequence[str]) -> Request: return self.b("PUT", "/me/tracks", params=dict(ids=",".join(ids))) class Auth(RequestBuilder): @returns(models.ClientCredentialsToken) def get_token_from_client_credentials(self) -> Request: """ Get an authorization token from client credentials using :py:attr:`~client.client_id` and :py:attr:`~client.client_secret` """ return self.b( "POST", self.client.token_url, params={"grant_type": "client_credentials"}, headers={ "Authorization": _basic_auth_str( # type: ignore self.client.client_id, self.client.client_secret ), "Content-Type": "application/x-www-form-urlencoded", }, ) @returns(models.UserToken) def get_token_from_refresh_token( self, refresh_token: str = None ) -> Request: """ Get a fresh authorization token from a refresh token :param refresh_token: A Spotify refresh token, if `None` will use the refresh token stored on client instance. """ if refresh_token: refresh = refresh_token elif self.client.token and self.client.token.refresh_token: refresh = self.client.token.refresh_token else: raise ValueError( "No refresh token passed and no token on instance" ) return self.b( "POST", self.client.token_url, params=dict(refresh_token=refresh, grant_type="refresh_token"), headers={ "Authorization": _basic_auth_str( # type: ignore self.client.client_id, self.client.client_secret ), "Content-Type": "application/x-www-form-urlencoded", }, ) @returns(models.UserToken) def get_token_from_code(self, response_code: str, **kwargs) -> Request: """ Get an authorization token from an OAuth response code :param response_code: The response code from redirect URL :param \\**kwargs: Additional keyword arguments that override instance attributes and are sent to the token API as query params """ params = dict( redirect_uri=self.client.redirect_uri, code=response_code, grant_type="authorization_code", scope=self.client.scope, state=self.client.state, ) params.update(kwargs) return self.b( "POST", self.client.token_url, params=params, headers={ "Authorization": _basic_auth_str( # type: ignore self.client.client_id, self.client.client_secret ), "Content-Type": "application/x-www-form-urlencoded", }, ) def _clear_nones(params: dict): return {key: value for key, value in params.items() if value is not None} def _basic_auth_str(username: str, password: str, prefix="Basic ") -> str: """ Create a base64 basic auth header value from username/password """ return ( prefix + b64encode(b":".join((username.encode(), password.encode()))).decode() ) PK!vspoffy/sansio.pyimport json from typing import MutableMapping, Union, Optional, Type, Any from urlobject import URLObject from spoffy.exceptions import ( SpotifyException, SpotifyUnauthorized, SpotifyPremiumRequired, ) class Request: """ :param method: The request method (`"GET"/"POST"/"PUT"/"DELETE"`) :param url: The request URL (absolute or relative to client base URL) :param body: Optional request body, can be passed either as raw bytes or in dict format (in which case it will be json encoded) :param headers: Additional request headers :param access_token: Will be added to Authorization header """ def __init__( self, method: str, url: str, params: Optional[MutableMapping] = None, body: Optional[Union[bytes, MutableMapping[str, Any]]] = None, headers: Optional[MutableMapping[str, str]] = None, access_token: Optional[str] = None, ): self.method = method self.url = str(URLObject(url).add_query_params(**(params or {}))) self.headers = dict(headers or {}) self.body = body if body is not None and isinstance(body, MutableMapping): charset = "utf-8" self.body = json.dumps(body).encode(charset) self.headers[ "Content-Type" ] = f"application/json; charset={charset}" self.headers["Content-Length"] = str(len(self.body)) elif body is not None and isinstance(body, bytes): self.body = body self.headers["Content-Length"] = str(len(self.body)) if access_token: self.headers["Authorization"] = "Bearer " + access_token def __repr__(self): return "<{}(method={}, url={}, body={}, headers={})>".format( self.__class__.__name__, repr(self.method), repr(self.url), repr(self.body), repr(self.headers), ) def __str__(self): return "Request: {} {}".format(self.method, self.url) class Response: def __init__( self, request: Request, status_code: int, headers: MutableMapping, content: Optional[bytes] = None, ): self.request = request self.status_code = status_code self.headers = headers self.content = content def raise_for_status(self): """ Raise a :class:`~spotify.exceptions.SpotifyException` if response status code is an error code. """ if self.status_code < 400: return kwargs = dict( status_code=self.status_code, headers=self.headers, request_method=self.request.method, request_url=self.request.url, ) try: error_info = self.json if isinstance(error_info["error"], dict): error_info = error_info["error"] if "error" in error_info and "message" not in error_info: error_info["message"] = error_info["error"] except Exception: error_info = {"status": self.status_code, "message": self.text} reason = error_info.get("reason") exc_class: Type[SpotifyException] if kwargs["status_code"] == 401: exc_class = SpotifyUnauthorized elif kwargs["status_code"] == 403 and reason == "PREMIUM_REQUIRED": exc_class = SpotifyPremiumRequired else: exc_class = SpotifyException kwargs["reason"] = reason kwargs["error_description"] = error_info.get("error_description") raise exc_class(error_info["message"], **kwargs) # type: ignore @property def json(self) -> Optional[dict]: if not self.content: return None return json.loads(self.content) @property def text(self) -> Optional[str]: if self.content is None: return None return self.content.decode() PK!3spoffy/spotify.pyfrom spoffy.client.base import AsyncClient, SyncClient from spoffy.modules.modules import ( Auth, AsyncAuth, Artists, AsyncArtists, Player, AsyncPlayer, Tracks, AsyncTracks, Albums, AsyncAlbums, Playlists, AsyncPlaylists, Search, AsyncSearch, ) class AsyncSpotify: """ Async Spotify API wrapper. Wraps the client object and exposes the Spotify web API organized into modules. :ivar ~Spotify.client: The underlying client instance :vartype ~Spotify.client: :py:class:`~AsyncClient` :ivar ~Spotify.auth: Authorization methods :vartype ~Spotify.auth: :py:class:`~spoffy.modules.modules.AsyncAuth` :ivar ~Spotify.albums: Access to album endpoints :vartype ~Spotify.albums: :py:class:`~spoffy.modules.modules.AsyncAlbums` :ivar ~Spotify.tracks: Access to track endpoints :vartype ~Spotify.tracks: :py:class:`~spoffy.modules.modules.AsyncTracks` :ivar ~Spotify.playlists: Access to playlist endpoints :vartype ~Spotify.playlists: :py:class:`~spoffy.modules.modules.AsyncPlaylists` :ivar ~Spotify.artists: Access to artist endpoints :vartype ~Spotify.artists: :py:class:`~spoffy.modules.modules.AsyncArtists` :ivar ~Spotify.search: Access to search endpoints :vartype ~Spotify.search: :py:class:`~spoffy.modules.modules.AsyncSearch` :ivar ~Spotify.player: Access to player endpoints :vartype ~Spotify.player: :py:class:`~spoffy.modules.modules.AsyncPlayer` """ # noqa def __init__(self, client: AsyncClient) -> None: """ :param client: An async client instance """ self.client = client self.artists = AsyncArtists(self.client) self.auth = AsyncAuth(self.client) self.player = AsyncPlayer(self.client) self.tracks = AsyncTracks(self.client) self.albums = AsyncAlbums(self.client) self.playlists = AsyncPlaylists(self.client) self.search = AsyncSearch(self.client) class SyncSpotify: """ Synchronous Spotify API wrapper. Wraps the client object and exposes the Spotify web API organized into modules. :ivar ~Spotify.client: The underlying client instance :vartype ~Spotify.client: :py:class:`~SyncClient` :ivar ~Spotify.auth: Authorization methods :vartype ~Spotify.auth: :py:class:`~spoffy.modules.modules.Auth` :ivar ~Spotify.albums: Access to album endpoints :vartype ~Spotify.albums: :py:class:`~spoffy.modules.modules.Albums` :ivar ~Spotify.tracks: Access to track endpoints :vartype ~Spotify.tracks: :py:class:`~spoffy.modules.modules.Tracks` :ivar ~Spotify.playlists: Access to playlist endpoints :vartype ~Spotify.playlists: :py:class:`~spoffy.modules.modules.Playlists` :ivar ~Spotify.artists: Access to artist endpoints :vartype ~Spotify.artists: :py:class:`~spoffy.modules.modules.Artists` :ivar ~Spotify.search: Access to search endpoints :vartype ~Spotify.search: :py:class:`~spoffy.modules.modules.Search` :ivar ~Spotify.player: Access to player endpoints :vartype ~Spotify.player: :py:class:`~spoffy.modules.modules.Player` """ def __init__(self, client: SyncClient) -> None: """ :param client: An sync client instance """ self.client = client self.artists = Artists(self.client) self.auth = Auth(self.client) self.player = Player(self.client) self.tracks = Tracks(self.client) self.albums = Albums(self.client) self.playlists = Playlists(self.client) self.search = Search(self.client) PK!P5---spoffy-0.1.0.dist-info/LICENSECopyright 2019 Steinthor Palsson Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.PK!HڽTUspoffy-0.1.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HW]Nlspoffy-0.1.0.dist-info/METADATAOK@)h!YG* iAТ& oH=foQzܣEZGFUd7M3'l7(v,`ODGR{$tzGᆾ7ֳ;dcbR_Ɔh/IOj3ۉBBɧErN3FJc C2 o:t!6!4"g싫|upZ̒__"?3μK',|ƾPK!Hw ſ spoffy-0.1.0.dist-info/RECORDɲFཟ]3 /@ČĆ(1ψAq5ֹt"̬KC&{M>n ǩFKs!H}jS6T5+t b~+U7@: FWgC`˵w |<Yքvg?a5p0;$pS0Úrd -Ʉ  &?qJvR.FP#xv:s$^ ֋#apuMt~air2dx6fS-yZ-"ԗ/a܍y ; a9Db7%EZn '4{}T !}ayFƱI}aƺ>Xԏ[b)$sJi5FVX#wnq!d{BZVU~^9"D :mlx,#ĝs{bs~hPosJ*OKHr>2u-v`1:[4= __ 8y;pTvk©&d!ƶ+}˪q(&\[.5[ȟ@}z8Aȁqal.#'Y n"!Gn2@{Z{Z1pR%5v(`LG[ SZyi#f&RSC"Sk*9JJ4vۚU2& W%h[] aM!o$@G>s8SW,gJ8ٞ-ȿߨoRgOS9`&i %b D,egϘ'RŋsI QOQ$"`z1U*W#ڶ7I"f _,5ѤQzj `XF:pS.¶AI|/OhMyʯ[/ăVXod[+<ѥhHU#k?_n~{vtgMz7ΰE4Pkl|+ƚ5PK! spoffy/__init__.pyPK!spoffy/client/__init__.pyPK!n&&Fspoffy/client/base.pyPK!0ӣjspoffy/client/common.pyPK!spoffy/clientmodule.pyPK!ͱspoffy/exceptions.pyPK!spoffy/io/__init__.pyPK![ht t spoffy/io/aiohttp.pyPK!rю 'spoffy/io/requests.pyPK!^rrX1spoffy/models/__init__.pyPK!-,87spoffy/models/audiofeatures.pyPK!Qwff8spoffy/models/base.pyPK!^w.[[Aspoffy/models/collections.pyPK!\HCspoffy/models/compat.pyPK!Z= = FGspoffy/models/core.pyPK!B蕡Pspoffy/models/image.pyPK!zQspoffy/models/library.pyPK!#Sspoffy/models/paging.pyPK!Շ iXspoffy/models/personalization.pyPK!hRsYspoffy/models/player.pyPK!>Zˋ{{]spoffy/models/playlists.pyPK!+9espoffy/models/search.pyPK!#bbahspoffy/models/token.pyPK! **mspoffy/models/users.pyPK!Upspoffy/modules/__init__.pyPK!P]mmpspoffy/modules/apimodule.pyPK!URHH3uspoffy/modules/mixins.pyPK!p|p|wspoffy/modules/modules.pyPK!¤DDXspoffy/modules/sansio.pyPK!vb9spoffy/sansio.pyPK!3Ispoffy/spotify.pyPK!P5---bWspoffy-0.1.0.dist-info/LICENSEPK!HڽTUYspoffy-0.1.0.dist-info/WHEELPK!HW]NlYZspoffy-0.1.0.dist-info/METADATAPK!Hw ſ [spoffy-0.1.0.dist-info/RECORDPK## a