diff --git a/spotipy/cache_handler.py b/spotipy/cache_handler.py index babfb21..5bd0a10 100644 --- a/spotipy/cache_handler.py +++ b/spotipy/cache_handler.py @@ -11,6 +11,8 @@ import errno import json import logging import os +from json import JSONEncoder +from typing import Optional from redis import RedisError @@ -49,10 +51,12 @@ class CacheFileHandler(CacheHandler): as json files on disk. """ - def __init__(self, - cache_path=None, - username=None, - encoder_cls=None): + def __init__( + self, + cache_path: Optional[str] = None, + username: Optional[str] = None, + encoder_cls: Optional[JSONEncoder] = None, + ): """ Parameters: * cache_path: May be supplied, will otherwise be generated @@ -185,7 +189,7 @@ class RedisCacheHandler(CacheHandler): A cache handler that stores the token info in the Redis. """ - def __init__(self, redis, key=None): + def __init__(self, redis, key: Optional[str] = None): """ Parameters: * redis: Redis object provided by redis-py library @@ -194,7 +198,7 @@ class RedisCacheHandler(CacheHandler): (takes precedence over `token_info`) """ self.redis = redis - self.key = key if key else 'token_info' + self.key: str = key if key else 'token_info' def get_cached_token(self): token_info = None @@ -218,7 +222,7 @@ class MemcacheCacheHandler(CacheHandler): """A Cache handler that stores the token info in Memcache using the pymemcache client """ - def __init__(self, memcache, key=None) -> None: + def __init__(self, memcache, key: Optional[str] = None): """ Parameters: * memcache: memcache client object provided by pymemcache @@ -227,7 +231,7 @@ class MemcacheCacheHandler(CacheHandler): (takes precedence over `token_info`) """ self.memcache = memcache - self.key = key if key else 'token_info' + self.key: str = key if key else 'token_info' def get_cached_token(self): from pymemcache import MemcacheError diff --git a/spotipy/client.py b/spotipy/client.py index 92fe1da..5b9cfdb 100644 --- a/spotipy/client.py +++ b/spotipy/client.py @@ -7,11 +7,12 @@ import logging import re import warnings from collections import defaultdict +from typing import Optional, Union, List, Dict import requests from spotipy.exceptions import SpotifyException -from spotipy.util import REQUESTS_SESSION, Retry +from spotipy.util import REQUESTS_SESSION, Retry, StrListOrTuple logger = logging.getLogger(__name__) @@ -130,12 +131,12 @@ class Spotify: oauth_manager=None, auth_manager=None, proxies=None, - requests_timeout=5, + requests_timeout: int = 5, status_forcelist=None, - retries=max_retries, - status_retries=max_retries, - backoff_factor=0.3, - language=None, + retries: int = max_retries, + status_retries: int = max_retries, + backoff_factor: float = 0.3, + language: Optional[str] = None, ): """ Creates a Spotify API client. @@ -240,7 +241,7 @@ class Spotify: token = self.auth_manager.get_access_token() return {"Authorization": f"Bearer {token}"} - def _internal_call(self, method, url, payload, params): + def _internal_call(self, method: str, url: str, payload, params): args = dict(params=params) if not url.startswith("http"): url = self.prefix + url @@ -314,28 +315,28 @@ class Spotify: logger.debug(f'RESULTS: {results}') return results - def _get(self, url, args=None, payload=None, **kwargs): + def _get(self, url: str, args=None, payload=None, **kwargs): if args: kwargs.update(args) return self._internal_call("GET", url, payload, kwargs) - def _post(self, url, args=None, payload=None, **kwargs): + def _post(self, url: str, args=None, payload=None, **kwargs): if args: kwargs.update(args) return self._internal_call("POST", url, payload, kwargs) - def _delete(self, url, args=None, payload=None, **kwargs): + def _delete(self, url: str, args=None, payload=None, **kwargs): if args: kwargs.update(args) return self._internal_call("DELETE", url, payload, kwargs) - def _put(self, url, args=None, payload=None, **kwargs): + def _put(self, url: str, args=None, payload=None, **kwargs): if args: kwargs.update(args) return self._internal_call("PUT", url, payload, kwargs) - def next(self, result): + def next(self, result: Dict): """ returns the next result given a paged result Parameters: @@ -346,7 +347,7 @@ class Spotify: else: return None - def previous(self, result): + def previous(self, result: Dict): """ returns the previous result given a paged result Parameters: @@ -357,7 +358,7 @@ class Spotify: else: return None - def track(self, track_id, market=None): + def track(self, track_id: str, market: Optional[str] = None): """ returns a single track given the track's ID, URI or URL Parameters: @@ -368,7 +369,7 @@ class Spotify: trid = self._get_id("track", track_id) return self._get("tracks/" + trid, market=market) - def tracks(self, tracks, market=None): + def tracks(self, tracks: StrListOrTuple, market: Optional[str] = None): """ returns a list of tracks given a list of track IDs, URIs, or URLs Parameters: @@ -379,7 +380,7 @@ class Spotify: tlist = [self._get_id("track", t) for t in tracks] return self._get("tracks/?ids=" + ",".join(tlist), market=market) - def artist(self, artist_id): + def artist(self, artist_id: str): """ returns a single artist given the artist's ID, URI or URL Parameters: @@ -389,7 +390,7 @@ class Spotify: trid = self._get_id("artist", artist_id) return self._get("artists/" + trid) - def artists(self, artists): + def artists(self, artists: StrListOrTuple): """ returns a list of artists given the artist IDs, URIs, or URLs Parameters: @@ -400,7 +401,13 @@ class Spotify: return self._get("artists/?ids=" + ",".join(tlist)) def artist_albums( - self, artist_id, album_type=None, include_groups=None, country=None, limit=20, offset=0 + self, + artist_id: str, + album_type=None, + include_groups=None, + country=None, + limit: int = 20, + offset: int = 0, ): """ Get Spotify catalog information about an artist's albums @@ -436,7 +443,7 @@ class Spotify: offset=offset, ) - def artist_top_tracks(self, artist_id, country="US"): + def artist_top_tracks(self, artist_id: str, country: str = "US"): """ Get Spotify catalog information about an artist's top 10 tracks by country. @@ -448,7 +455,7 @@ class Spotify: trid = self._get_id("artist", artist_id) return self._get("artists/" + trid + "/top-tracks", country=country) - def artist_related_artists(self, artist_id): + def artist_related_artists(self, artist_id: str): """ Get Spotify catalog information about artists similar to an identified artist. Similarity is based on analysis of the Spotify community's listening history. @@ -467,7 +474,7 @@ class Spotify: trid = self._get_id("artist", artist_id) return self._get("artists/" + trid + "/related-artists") - def album(self, album_id, market=None): + def album(self, album_id: str, market: Optional[str] = None): """ returns a single album given the album's ID, URIs or URL Parameters: @@ -481,7 +488,13 @@ class Spotify: else: return self._get("albums/" + trid) - def album_tracks(self, album_id, limit=50, offset=0, market=None): + def album_tracks( + self, + album_id: str, + limit: int = 50, + offset: int = 0, + market: Optional[str] = None, + ): """ Get Spotify catalog information about an album's tracks Parameters: @@ -497,7 +510,7 @@ class Spotify: "albums/" + trid + "/tracks/", limit=limit, offset=offset, market=market ) - def albums(self, albums, market=None): + def albums(self, albums: StrListOrTuple, market: Optional[str] = None): """ returns a list of albums given the album IDs, URIs, or URLs Parameters: @@ -511,7 +524,7 @@ class Spotify: else: return self._get("albums/?ids=" + ",".join(tlist)) - def show(self, show_id, market=None): + def show(self, show_id: str, market: Optional[str] = None): """ returns a single show given the show's ID, URIs or URL Parameters: @@ -526,7 +539,7 @@ class Spotify: trid = self._get_id("show", show_id) return self._get("shows/" + trid, market=market) - def shows(self, shows, market=None): + def shows(self, shows: StrListOrTuple, market: Optional[str] = None): """ returns a list of shows given the show IDs, URIs, or URLs Parameters: @@ -541,7 +554,13 @@ class Spotify: tlist = [self._get_id("show", s) for s in shows] return self._get("shows/?ids=" + ",".join(tlist), market=market) - def show_episodes(self, show_id, limit=50, offset=0, market=None): + def show_episodes( + self, + show_id: str, + limit: int = 50, + offset: int = 0, + market: Optional[str] = None, + ): """ Get Spotify catalog information about a show's episodes Parameters: @@ -560,7 +579,7 @@ class Spotify: "shows/" + trid + "/episodes/", limit=limit, offset=offset, market=market ) - def episode(self, episode_id, market=None): + def episode(self, episode_id: str, market: Optional[str] = None): """ returns a single episode given the episode's ID, URIs or URL Parameters: @@ -575,7 +594,7 @@ class Spotify: trid = self._get_id("episode", episode_id) return self._get("episodes/" + trid, market=market) - def episodes(self, episodes, market=None): + def episodes(self, episodes: StrListOrTuple, market: Optional[str] = None): """ returns a list of episodes given the episode IDs, URIs, or URLs Parameters: @@ -590,7 +609,14 @@ class Spotify: tlist = [self._get_id("episode", e) for e in episodes] return self._get("episodes/?ids=" + ",".join(tlist), market=market) - def search(self, q, limit=10, offset=0, type="track", market=None): + def search( + self, + q: str, + limit: int = 10, + offset: int = 0, + type: str = "track", + market: Optional[str] = None, + ): """ searches for an item Parameters: @@ -609,7 +635,15 @@ class Spotify: "search", q=q, limit=limit, offset=offset, type=type, market=market ) - def search_markets(self, q, limit=10, offset=0, type="track", markets=None, total=None): + def search_markets( + self, + q: str, + limit: int = 10, + offset: int = 0, + type: str = "track", + markets: Optional[StrListOrTuple] = None, + total: Optional[int] = None, + ): """ (experimental) Searches multiple markets for an item Parameters: @@ -641,7 +675,7 @@ class Spotify: ) return self._search_multiple_markets(q, limit, offset, type, markets, total) - def user(self, user): + def user(self, user: str): """ Gets basic profile information about a Spotify User Parameters: @@ -649,7 +683,7 @@ class Spotify: """ return self._get("users/" + user) - def current_user_playlists(self, limit=50, offset=0): + def current_user_playlists(self, limit: int = 50, offset: int = 0): """ Get current user playlists without required getting his profile Parameters: - limit - the number of items to return @@ -657,7 +691,13 @@ class Spotify: """ return self._get("me/playlists", limit=limit, offset=offset) - def playlist(self, playlist_id, fields=None, market=None, additional_types=("track",)): + def playlist( + self, + playlist_id: str, + fields=None, + market: Optional[str] = None, + additional_types: StrListOrTuple = ("track",), + ): """ Gets playlist by id. Parameters: @@ -678,12 +718,12 @@ class Spotify: def playlist_tracks( self, - playlist_id, + playlist_id: str, fields=None, - limit=100, - offset=0, - market=None, - additional_types=("track",) + limit: int = 100, + offset: int = 0, + market: Optional[str] = None, + additional_types: StrListOrTuple = ("track",) ): """ Get full details of the tracks of a playlist. @@ -712,10 +752,10 @@ class Spotify: self, playlist_id, fields=None, - limit=100, - offset=0, - market=None, - additional_types=("track", "episode") + limit: int = 100, + offset: int = 0, + market: Optional[str] = None, + additional_types: StrListOrTuple = ("track", "episode") ): """ Get full details of the tracks and episodes of a playlist. @@ -738,7 +778,7 @@ class Spotify: additional_types=",".join(additional_types) ) - def playlist_cover_image(self, playlist_id): + def playlist_cover_image(self, playlist_id: str): """ Get cover image of a playlist. Parameters: @@ -747,7 +787,7 @@ class Spotify: plid = self._get_id("playlist", playlist_id) return self._get(f"playlists/{plid}/images") - def playlist_upload_cover_image(self, playlist_id, image_b64): + def playlist_upload_cover_image(self, playlist_id: str, image_b64): """ Replace the image used to represent a specific playlist Parameters: @@ -762,7 +802,13 @@ class Spotify: content_type="image/jpeg", ) - def user_playlist(self, user, playlist_id=None, fields=None, market=None): + def user_playlist( + self, + user: str, + playlist_id: Optional[str] = None, + fields=None, + market: Optional[str] = None, + ): """ Gets a single playlist of a user .. deprecated:: @@ -785,12 +831,12 @@ class Spotify: def user_playlist_tracks( self, - user=None, - playlist_id=None, + user: Optional[str] = None, + playlist_id: Optional[str] = None, fields=None, - limit=100, - offset=0, - market=None, + limit: int = 100, + offset: int = 0, + market: Optional[str] = None, ): """ Get full details of the tracks of a playlist owned by a user. @@ -818,7 +864,7 @@ class Spotify: market=market, ) - def user_playlists(self, user, limit=50, offset=0): + def user_playlists(self, user: str, limit: int = 50, offset: int = 0): """ Gets playlists of a user Parameters: @@ -830,7 +876,14 @@ class Spotify: f"users/{user}/playlists", limit=limit, offset=offset ) - def user_playlist_create(self, user, name, public=True, collaborative=False, description=""): + def user_playlist_create( + self, + user: str, + name: str, + public: bool = True, + collaborative: bool = False, + description: str = "", + ): """ Creates a playlist for a user Parameters: @@ -851,12 +904,12 @@ class Spotify: def user_playlist_change_details( self, - user, - playlist_id, - name=None, - public=None, - collaborative=None, - description=None, + user: str, + playlist_id: str, + name: Optional[str] = None, + public: Optional[bool] = None, + collaborative: Optional[bool] = None, + description: Optional[str] = None, ): """ This function is no longer in use, please use the recommended function in the warning! @@ -882,7 +935,7 @@ class Spotify: return self.playlist_change_details(playlist_id, name, public, collaborative, description) - def user_playlist_unfollow(self, user, playlist_id): + def user_playlist_unfollow(self, user: str, playlist_id: str): """ This function is no longer in use, please use the recommended function in the warning! Unfollows (deletes) a playlist for a user @@ -902,7 +955,11 @@ class Spotify: return self.current_user_unfollow_playlist(playlist_id) def user_playlist_add_tracks( - self, user, playlist_id, tracks, position=None + self, + user: str, + playlist_id: str, + tracks: StrListOrTuple, + position: Optional[int] = None, ): """ This function is no longer in use, please use the recommended function in the warning! @@ -927,7 +984,11 @@ class Spotify: return self.playlist_add_items(playlist_id, tracks, position) def user_playlist_add_episodes( - self, user, playlist_id, episodes, position=None + self, + user: str, + playlist_id: str, + episodes: StrListOrTuple, + position: Optional[int] = None, ): """ This function is no longer in use, please use the recommended function in the warning! @@ -951,7 +1012,9 @@ class Spotify: episodes = [self._get_uri("episode", tid) for tid in episodes] return self.playlist_add_items(playlist_id, episodes, position) - def user_playlist_replace_tracks(self, user, playlist_id, tracks): + def user_playlist_replace_tracks( + self, user: str, playlist_id: str, tracks: StrListOrTuple + ): """ This function is no longer in use, please use the recommended function in the warning! Replace all tracks in a playlist for a user @@ -973,12 +1036,12 @@ class Spotify: def user_playlist_reorder_tracks( self, - user, - playlist_id, - range_start, - insert_before, - range_length=1, - snapshot_id=None, + user: str, + playlist_id: str, + range_start: int, + insert_before: int, + range_length: int = 1, + snapshot_id: Optional[str] = None, ): """ This function is no longer in use, please use the recommended function in the warning! @@ -1007,7 +1070,11 @@ class Spotify: snapshot_id) def user_playlist_remove_all_occurrences_of_tracks( - self, user, playlist_id, tracks, snapshot_id=None + self, + user: str, + playlist_id: str, + tracks: StrListOrTuple, + snapshot_id: Optional[str] = None, ): """ This function is no longer in use, please use the recommended function in the warning! @@ -1033,7 +1100,11 @@ class Spotify: snapshot_id) def user_playlist_remove_specific_occurrences_of_tracks( - self, user, playlist_id, tracks, snapshot_id=None + self, + user: str, + playlist_id: str, + tracks: List[Dict], + snapshot_id: Optional[str] = None, ): """ This function is no longer in use, please use the recommended function in the warning! @@ -1073,7 +1144,7 @@ class Spotify: f"users/{user}/playlists/{plid}/tracks", payload=payload ) - def user_playlist_follow_playlist(self, playlist_owner_id, playlist_id): + def user_playlist_follow_playlist(self, playlist_owner_id: str, playlist_id: str): """ This function is no longer in use, please use the recommended function in the warning! Add the current authenticated user as a follower of a playlist. @@ -1093,7 +1164,7 @@ class Spotify: return self.current_user_follow_playlist(playlist_id) def user_playlist_is_following( - self, playlist_owner_id, playlist_id, user_ids + self, playlist_owner_id: str, playlist_id: str, user_ids: StrListOrTuple ): """ This function is no longer in use, please use the recommended function in the warning! @@ -1117,11 +1188,11 @@ class Spotify: def playlist_change_details( self, - playlist_id, - name=None, - public=None, - collaborative=None, - description=None, + playlist_id: str, + name: Optional[str] = None, + public: Optional[bool] = None, + collaborative: Optional[bool] = None, + description: Optional[str] = None, ): """ Changes a playlist's name and/or public/private state, collaborative state, and/or description @@ -1147,7 +1218,7 @@ class Spotify: f"playlists/{self._get_id('playlist', playlist_id)}", payload=data ) - def current_user_unfollow_playlist(self, playlist_id): + def current_user_unfollow_playlist(self, playlist_id: str): """ Unfollows (deletes) a playlist for the current authenticated user @@ -1159,7 +1230,7 @@ class Spotify: ) def playlist_add_items( - self, playlist_id, items, position=None + self, playlist_id: str, items: StrListOrTuple, position: Optional[int] = None ): """ Adds tracks/episodes to a playlist @@ -1176,7 +1247,7 @@ class Spotify: position=position, ) - def playlist_replace_items(self, playlist_id, items): + def playlist_replace_items(self, playlist_id: str, items: StrListOrTuple): """ Replace all tracks/episodes in a playlist Parameters: @@ -1192,11 +1263,11 @@ class Spotify: def playlist_reorder_items( self, - playlist_id, - range_start, - insert_before, - range_length=1, - snapshot_id=None, + playlist_id: str, + range_start: int, + insert_before: int, + range_length: int = 1, + snapshot_id: Optional[str] = None, ): """ Reorder tracks in a playlist @@ -1222,7 +1293,7 @@ class Spotify: ) def playlist_remove_all_occurrences_of_items( - self, playlist_id, items, snapshot_id=None + self, playlist_id: str, items: StrListOrTuple, snapshot_id: Optional[str] = None ): """ Removes all occurrences of the given tracks/episodes from the given playlist @@ -1243,7 +1314,7 @@ class Spotify: ) def playlist_remove_specific_occurrences_of_items( - self, playlist_id, items, snapshot_id=None + self, playlist_id: str, items: List[Dict], snapshot_id: Optional[str] = None ): """ Removes all occurrences of the given tracks from the given playlist @@ -1273,7 +1344,7 @@ class Spotify: f"playlists/{plid}/tracks", payload=payload ) - def current_user_follow_playlist(self, playlist_id, public=True): + def current_user_follow_playlist(self, playlist_id: str, public: bool = True): """ Add the current authenticated user as a follower of a playlist. @@ -1287,7 +1358,7 @@ class Spotify: ) def playlist_is_following( - self, playlist_id, user_ids + self, playlist_id: str, user_ids: StrListOrTuple ): """ Check to see if the given users are following the given playlist @@ -1313,7 +1384,11 @@ class Spotify: """ return self.me() - def current_user_playing_track(self, market=None, additional_types=("track",)): + def current_user_playing_track( + self, + market: Optional[str] = None, + additional_types: StrListOrTuple = ("track",) + ): """ Get information about the current users currently playing track. Parameters: @@ -1328,7 +1403,9 @@ class Spotify: additional_types=",".join(additional_types) ) - def current_user_saved_albums(self, limit=20, offset=0, market=None): + def current_user_saved_albums( + self, limit: int = 20, offset: int = 0, market: Optional[str] = None + ): """ Gets a list of the albums saved in the current authorized user's "Your Music" library @@ -1340,7 +1417,7 @@ class Spotify: """ return self._get("me/albums", limit=limit, offset=offset, market=market) - def current_user_saved_albums_add(self, albums=[]): + def current_user_saved_albums_add(self, albums: StrListOrTuple = []): """ Add one or more albums to the current user's "Your Music" library. Parameters: @@ -1350,7 +1427,7 @@ class Spotify: alist = [self._get_id("album", a) for a in albums] return self._put("me/albums?ids=" + ",".join(alist)) - def current_user_saved_albums_delete(self, albums=[]): + def current_user_saved_albums_delete(self, albums: StrListOrTuple = []): """ Remove one or more albums from the current user's "Your Music" library. @@ -1360,7 +1437,7 @@ class Spotify: alist = [self._get_id("album", a) for a in albums] return self._delete("me/albums/?ids=" + ",".join(alist)) - def current_user_saved_albums_contains(self, albums=[]): + def current_user_saved_albums_contains(self, albums: StrListOrTuple = []): """ Check if one or more albums is already saved in the current Spotify user’s “Your Music” library. @@ -1370,7 +1447,9 @@ class Spotify: alist = [self._get_id("album", a) for a in albums] return self._get("me/albums/contains?ids=" + ",".join(alist)) - def current_user_saved_tracks(self, limit=20, offset=0, market=None): + def current_user_saved_tracks( + self, limit: int = 20, offset: int = 0, market: Optional[str] = None + ): """ Gets a list of the tracks saved in the current authorized user's "Your Music" library @@ -1382,43 +1461,39 @@ class Spotify: """ return self._get("me/tracks", limit=limit, offset=offset, market=market) - def current_user_saved_tracks_add(self, tracks=None): + def current_user_saved_tracks_add(self, tracks: StrListOrTuple = []): """ Add one or more tracks to the current user's "Your Music" library. Parameters: - tracks - a list of track URIs, URLs or IDs """ - tlist = [] - if tracks is not None: - tlist = [self._get_id("track", t) for t in tracks] + tlist = [self._get_id("track", t) for t in tracks] return self._put("me/tracks/?ids=" + ",".join(tlist)) - def current_user_saved_tracks_delete(self, tracks=None): + def current_user_saved_tracks_delete(self, tracks: StrListOrTuple = []): """ Remove one or more tracks from the current user's "Your Music" library. Parameters: - tracks - a list of track URIs, URLs or IDs """ - tlist = [] - if tracks is not None: - tlist = [self._get_id("track", t) for t in tracks] + tlist = [self._get_id("track", t) for t in tracks] return self._delete("me/tracks/?ids=" + ",".join(tlist)) - def current_user_saved_tracks_contains(self, tracks=None): + def current_user_saved_tracks_contains(self, tracks: StrListOrTuple = []): """ Check if one or more tracks is already saved in the current Spotify user’s “Your Music” library. Parameters: - tracks - a list of track URIs, URLs or IDs """ - tlist = [] - if tracks is not None: - tlist = [self._get_id("track", t) for t in tracks] + tlist = [self._get_id("track", t) for t in tracks] return self._get("me/tracks/contains?ids=" + ",".join(tlist)) - def current_user_saved_episodes(self, limit=20, offset=0, market=None): + def current_user_saved_episodes( + self, limit: int = 20, offset: int = 0, market: Optional[str] = None + ): """ Gets a list of the episodes saved in the current authorized user's "Your Music" library @@ -1430,43 +1505,39 @@ class Spotify: """ return self._get("me/episodes", limit=limit, offset=offset, market=market) - def current_user_saved_episodes_add(self, episodes=None): + def current_user_saved_episodes_add(self, episodes: StrListOrTuple = []): """ Add one or more episodes to the current user's "Your Music" library. Parameters: - episodes - a list of episode URIs, URLs or IDs """ - elist = [] - if episodes is not None: - elist = [self._get_id("episode", e) for e in episodes] + elist = [self._get_id("episode", e) for e in episodes] return self._put("me/episodes/?ids=" + ",".join(elist)) - def current_user_saved_episodes_delete(self, episodes=None): + def current_user_saved_episodes_delete(self, episodes: StrListOrTuple = []): """ Remove one or more episodes from the current user's "Your Music" library. Parameters: - episodes - a list of episode URIs, URLs or IDs """ - elist = [] - if episodes is not None: - elist = [self._get_id("episode", e) for e in episodes] + elist = [self._get_id("episode", e) for e in episodes] return self._delete("me/episodes/?ids=" + ",".join(elist)) - def current_user_saved_episodes_contains(self, episodes=None): + def current_user_saved_episodes_contains(self, episodes: StrListOrTuple = []): """ Check if one or more episodes is already saved in the current Spotify user’s “Your Music” library. Parameters: - episodes - a list of episode URIs, URLs or IDs """ - elist = [] - if episodes is not None: - elist = [self._get_id("episode", e) for e in episodes] + elist = [self._get_id("episode", e) for e in episodes] return self._get("me/episodes/contains?ids=" + ",".join(elist)) - def current_user_saved_shows(self, limit=20, offset=0, market=None): + def current_user_saved_shows( + self, limit: int = 20, offset: int = 0, market: Optional[str] = None + ): """ Gets a list of the shows saved in the current authorized user's "Your Music" library @@ -1478,7 +1549,7 @@ class Spotify: """ return self._get("me/shows", limit=limit, offset=offset, market=market) - def current_user_saved_shows_add(self, shows=[]): + def current_user_saved_shows_add(self, shows: StrListOrTuple = []): """ Add one or more albums to the current user's "Your Music" library. Parameters: @@ -1487,7 +1558,7 @@ class Spotify: slist = [self._get_id("show", s) for s in shows] return self._put("me/shows?ids=" + ",".join(slist)) - def current_user_saved_shows_delete(self, shows=[]): + def current_user_saved_shows_delete(self, shows: StrListOrTuple = []): """ Remove one or more shows from the current user's "Your Music" library. @@ -1497,7 +1568,7 @@ class Spotify: slist = [self._get_id("show", s) for s in shows] return self._delete("me/shows/?ids=" + ",".join(slist)) - def current_user_saved_shows_contains(self, shows=[]): + def current_user_saved_shows_contains(self, shows: StrListOrTuple = []): """ Check if one or more shows is already saved in the current Spotify user’s “Your Music” library. @@ -1507,7 +1578,9 @@ class Spotify: slist = [self._get_id("show", s) for s in shows] return self._get("me/shows/contains?ids=" + ",".join(slist)) - def current_user_followed_artists(self, limit=20, after=None): + def current_user_followed_artists( + self, limit: int = 20, after: Optional[str] = None + ): """ Gets a list of the artists followed by the current authorized user Parameters: @@ -1520,7 +1593,7 @@ class Spotify: "me/following", type="artist", limit=limit, after=after ) - def current_user_following_artists(self, ids=None): + def current_user_following_artists(self, ids: StrListOrTuple = []): """ Check if the current user is following certain artists Returns list of booleans respective to ids @@ -1528,14 +1601,12 @@ class Spotify: Parameters: - ids - a list of artist URIs, URLs or IDs """ - idlist = [] - if ids is not None: - idlist = [self._get_id("artist", i) for i in ids] + idlist = [self._get_id("artist", i) for i in ids] return self._get( "me/following/contains", ids=",".join(idlist), type="artist" ) - def current_user_following_users(self, ids=None): + def current_user_following_users(self, ids: StrListOrTuple = []): """ Check if the current user is following certain users Returns list of booleans respective to ids @@ -1543,15 +1614,13 @@ class Spotify: Parameters: - ids - a list of user URIs, URLs or IDs """ - idlist = [] - if ids is not None: - idlist = [self._get_id("user", i) for i in ids] + idlist = [self._get_id("user", i) for i in ids] return self._get( "me/following/contains", ids=",".join(idlist), type="user" ) def current_user_top_artists( - self, limit=20, offset=0, time_range="medium_term" + self, limit: int = 20, offset: int = 0, time_range: str = "medium_term" ): """ Get the current user's top artists @@ -1566,7 +1635,7 @@ class Spotify: ) def current_user_top_tracks( - self, limit=20, offset=0, time_range="medium_term" + self, limit: int = 20, offset: int = 0, time_range: str = "medium_term" ): """ Get the current user's top tracks @@ -1580,7 +1649,9 @@ class Spotify: "me/top/tracks", time_range=time_range, limit=limit, offset=offset ) - def current_user_recently_played(self, limit=50, after=None, before=None): + def current_user_recently_played( + self, limit: int = 50, after: Optional[int] = None, before: Optional[int] = None + ): """ Get the current user's recently played tracks Parameters: @@ -1599,28 +1670,28 @@ class Spotify: before=before, ) - def user_follow_artists(self, ids=[]): + def user_follow_artists(self, ids: StrListOrTuple = []): """ Follow one or more artists Parameters: - ids - a list of artist IDs """ return self._put("me/following?type=artist&ids=" + ",".join(ids)) - def user_follow_users(self, ids=[]): + def user_follow_users(self, ids: StrListOrTuple = []): """ Follow one or more users Parameters: - ids - a list of user IDs """ return self._put("me/following?type=user&ids=" + ",".join(ids)) - def user_unfollow_artists(self, ids=[]): + def user_unfollow_artists(self, ids: StrListOrTuple = []): """ Unfollow one or more artists Parameters: - ids - a list of artist IDs """ return self._delete("me/following?type=artist&ids=" + ",".join(ids)) - def user_unfollow_users(self, ids=[]): + def user_unfollow_users(self, ids: StrListOrTuple = []): """ Unfollow one or more users Parameters: - ids - a list of user IDs @@ -1628,7 +1699,12 @@ class Spotify: return self._delete("me/following?type=user&ids=" + ",".join(ids)) def featured_playlists( - self, locale=None, country=None, timestamp=None, limit=20, offset=0 + self, + locale: Optional[str] = None, + country: Optional[str] = None, + timestamp: Optional[str] = None, + limit: int = 20, + offset: int = 0, ): """ Get a list of Spotify featured playlists @@ -1668,7 +1744,9 @@ class Spotify: offset=offset, ) - def new_releases(self, country=None, limit=20, offset=0): + def new_releases( + self, country: Optional[str] = None, limit: int = 20, offset: int = 0 + ): """ Get a list of new album releases featured in Spotify Parameters: @@ -1685,7 +1763,12 @@ class Spotify: "browse/new-releases", country=country, limit=limit, offset=offset ) - def category(self, category_id, country=None, locale=None): + def category( + self, + category_id: str, + country: Optional[str] = None, + locale: Optional[str] = None, + ): """ Get info about a category Parameters: @@ -1702,7 +1785,13 @@ class Spotify: locale=locale, ) - def categories(self, country=None, locale=None, limit=20, offset=0): + def categories( + self, + country: Optional[str] = None, + locale: Optional[str] = None, + limit: int = 20, + offset: int = 0, + ): """ Get a list of categories Parameters: @@ -1727,7 +1816,11 @@ class Spotify: ) def category_playlists( - self, category_id=None, country=None, limit=20, offset=0 + self, + category_id: Optional[str] = None, + country: Optional[str] = None, + limit: int = 20, + offset: int = 0, ): """ Get a list of playlists for a specific Spotify category @@ -1760,11 +1853,11 @@ class Spotify: def recommendations( self, - seed_artists=None, - seed_genres=None, - seed_tracks=None, - limit=20, - country=None, + seed_artists: Optional[StrListOrTuple] = None, + seed_genres: Optional[StrListOrTuple] = None, + seed_tracks: Optional[StrListOrTuple] = None, + limit: int = 20, + country: Optional[str] = None, **kwargs ): """ Get a list of recommended tracks for one to five seeds. @@ -1846,7 +1939,7 @@ class Spotify: ) return self._get("recommendations/available-genre-seeds") - def audio_analysis(self, track_id): + def audio_analysis(self, track_id: str): """ Get audio analysis for a track based upon its Spotify ID .. deprecated:: @@ -1863,7 +1956,7 @@ class Spotify: trid = self._get_id("track", track_id) return self._get("audio-analysis/" + trid) - def audio_features(self, tracks=[]): + def audio_features(self, tracks: Union[str, StrListOrTuple] = []): """ Get audio features for one or multiple tracks based upon their Spotify IDs .. deprecated:: @@ -1896,7 +1989,7 @@ class Spotify: """ return self._get("me/player/devices") - def current_playback(self, market=None, additional_types=None): + def current_playback(self, market: Optional[str] = None, additional_types=None): """ Get information about user's current playback. Parameters: @@ -1905,7 +1998,7 @@ class Spotify: """ return self._get("me/player", market=market, additional_types=additional_types) - def currently_playing(self, market=None, additional_types=None): + def currently_playing(self, market: Optional[str] = None, additional_types=None): """ Get user's currently playing track. Parameters: @@ -1915,7 +2008,7 @@ class Spotify: return self._get("me/player/currently-playing", market=market, additional_types=additional_types) - def transfer_playback(self, device_id, force_play=True): + def transfer_playback(self, device_id: str, force_play: bool = True): """ Transfer playback to another device. Note that the API accepts a list of device ids, but only actually supports one. @@ -1929,7 +2022,12 @@ class Spotify: return self._put("me/player", payload=data) def start_playback( - self, device_id=None, context_uri=None, uris=None, offset=None, position_ms=None + self, + device_id: Optional[str] = None, + context_uri: Optional[str] = None, + uris: Optional[List[str]] = None, + offset: Optional[Dict] = None, + position_ms: Optional[Union[int, float]] = None, ): """ Start or resume user's playback. @@ -1971,7 +2069,7 @@ class Spotify: self._append_device_id("me/player/play", device_id), payload=data ) - def pause_playback(self, device_id=None): + def pause_playback(self, device_id: Optional[str] = None): """ Pause user's playback. Parameters: @@ -1979,7 +2077,7 @@ class Spotify: """ return self._put(self._append_device_id("me/player/pause", device_id)) - def next_track(self, device_id=None): + def next_track(self, device_id: Optional[str] = None): """ Skip user's playback to next track. Parameters: @@ -1987,7 +2085,7 @@ class Spotify: """ return self._post(self._append_device_id("me/player/next", device_id)) - def previous_track(self, device_id=None): + def previous_track(self, device_id: Optional[str] = None): """ Skip user's playback to previous track. Parameters: @@ -1997,7 +2095,9 @@ class Spotify: self._append_device_id("me/player/previous", device_id) ) - def seek_track(self, position_ms, device_id=None): + def seek_track( + self, position_ms: Union[int, float], device_id: Optional[str] = None + ): """ Seek to position in current track. Parameters: @@ -2013,7 +2113,7 @@ class Spotify: ) ) - def repeat(self, state, device_id=None): + def repeat(self, state: str, device_id: Optional[str] = None): """ Set repeat mode for playback. Parameters: @@ -2029,7 +2129,7 @@ class Spotify: ) ) - def volume(self, volume_percent, device_id=None): + def volume(self, volume_percent: int, device_id: Optional[str] = None): """ Set playback volume. Parameters: @@ -2049,7 +2149,7 @@ class Spotify: ) ) - def shuffle(self, state, device_id=None): + def shuffle(self, state, device_id: Optional[str] = None): """ Toggle playback shuffling. Parameters: @@ -2070,7 +2170,7 @@ class Spotify: """ Gets the current user's queue """ return self._get("me/player/queue") - def add_to_queue(self, uri, device_id=None): + def add_to_queue(self, uri, device_id: Optional[str] = None): """ Adds a song to the end of a user's queue If device A is currently playing music, and you try to add to the queue @@ -2101,7 +2201,7 @@ class Spotify: """ return self._get("markets") - def _append_device_id(self, path, device_id): + def _append_device_id(self, path: str, device_id: Optional[str]): """ Append device ID to API path. Parameters: @@ -2114,7 +2214,7 @@ class Spotify: path += f"?device_id={device_id}" return path - def _get_id(self, type, id): + def _get_id(self, type, id: str): uri_match = re.search(Spotify._regex_spotify_uri, id) if uri_match is not None: uri_match_groups = uri_match.groupdict() @@ -2138,16 +2238,18 @@ class Spotify: # TODO change to a ValueError in v3 raise SpotifyException(400, -1, "Unsupported URL / URI.") - def _get_uri(self, type, id): + def _get_uri(self, type, id: str): if self._is_uri(id): return id else: return "spotify:" + type + ":" + self._get_id(type, id) - def _is_uri(self, uri): + def _is_uri(self, uri: str): return re.search(Spotify._regex_spotify_uri, uri) is not None - def _search_multiple_markets(self, q, limit, offset, type, markets, total): + def _search_multiple_markets( + self, q: str, limit: int, offset: int, type, markets: StrListOrTuple, total: int + ): if total and limit > total: limit = total warnings.warn(f"limit was auto-adjusted to equal {total} " @@ -2181,7 +2283,7 @@ class Spotify: return results - def get_audiobook(self, id, market=None): + def get_audiobook(self, id: str, market: Optional[str] = None): """ Get Spotify catalog information for a single audiobook identified by its unique Spotify ID. @@ -2197,7 +2299,7 @@ class Spotify: return self._get(endpoint) - def get_audiobooks(self, ids, market=None): + def get_audiobooks(self, ids: StrListOrTuple, market: Optional[str] = None): """ Get Spotify catalog information for multiple audiobooks based on their Spotify IDs. Parameters: @@ -2212,7 +2314,9 @@ class Spotify: return self._get(endpoint) - def get_audiobook_chapters(self, id, market=None, limit=20, offset=0): + def get_audiobook_chapters( + self, id: str, market: Optional[str] = None, limit: int = 20, offset: int = 0 + ): """ Get Spotify catalog information about an audiobook’s chapters. Parameters: diff --git a/spotipy/exceptions.py b/spotipy/exceptions.py index cacce9f..805f0d1 100644 --- a/spotipy/exceptions.py +++ b/spotipy/exceptions.py @@ -34,8 +34,16 @@ class SpotifyOauthError(SpotifyBaseException): class SpotifyStateError(SpotifyOauthError): """ The state sent and state received were different """ - def __init__(self, local_state=None, remote_state=None, message=None, - error=None, error_description=None, *args, **kwargs): + def __init__( + self, + local_state=None, + remote_state=None, + message=None, + error=None, + error_description=None, + *args, + **kwargs, + ): if not message: message = ("Expected " + local_state + " but received " + remote_state) diff --git a/spotipy/oauth2.py b/spotipy/oauth2.py index 60f00b9..970a9ea 100644 --- a/spotipy/oauth2.py +++ b/spotipy/oauth2.py @@ -16,6 +16,7 @@ import urllib.parse as urllibparse import warnings import webbrowser from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Optional, Union, Any from urllib.parse import parse_qsl, urlparse import requests @@ -23,19 +24,21 @@ import requests from spotipy.cache_handler import CacheFileHandler, CacheHandler from spotipy.exceptions import SpotifyOauthError, SpotifyStateError from spotipy.util import (CLIENT_CREDS_ENV_VARS, REQUESTS_SESSION, - get_host_port, normalize_scope) + get_host_port, normalize_scope, ScopeArgType) logger = logging.getLogger(__name__) +# TODO: improve the types that are "Any" -def _make_authorization_headers(client_id, client_secret): + +def _make_authorization_headers(client_id: str, client_secret: str): auth_header = base64.b64encode( str(client_id + ":" + client_secret).encode("ascii") ) return {"Authorization": f"Basic {auth_header.decode('ascii')}"} -def _ensure_value(value, env_key): +def _ensure_value(value: Optional[str], env_key: str) -> str: env_val = CLIENT_CREDS_ENV_VARS[env_key] _val = value or os.getenv(env_val) if _val is None: @@ -45,7 +48,7 @@ def _ensure_value(value, env_key): class SpotifyAuthBase: - def __init__(self, requests_session): + def __init__(self, requests_session: Optional[Union[requests.Session, bool]] = None): if isinstance(requests_session, requests.Session): self._session = requests_session else: @@ -55,7 +58,7 @@ class SpotifyAuthBase: from requests import api self._session = api - def _normalize_scope(self, scope): + def _normalize_scope(self, scope: Optional[ScopeArgType]): return normalize_scope(scope) @property @@ -63,7 +66,7 @@ class SpotifyAuthBase: return self._client_id @client_id.setter - def client_id(self, val): + def client_id(self, val: Optional[str]): self._client_id = _ensure_value(val, "client_id") @property @@ -71,7 +74,7 @@ class SpotifyAuthBase: return self._client_secret @client_secret.setter - def client_secret(self, val): + def client_secret(self, val: Optional[str]): self._client_secret = _ensure_value(val, "client_secret") @property @@ -79,30 +82,30 @@ class SpotifyAuthBase: return self._redirect_uri @redirect_uri.setter - def redirect_uri(self, val): + def redirect_uri(self, val: Optional[str]): self._redirect_uri = _ensure_value(val, "redirect_uri") @staticmethod - def _get_user_input(prompt): + def _get_user_input(prompt) -> str: try: return raw_input(prompt) except NameError: return input(prompt) @staticmethod - def is_token_expired(token_info): + def is_token_expired(token_info) -> bool: now = int(time.time()) return token_info["expires_at"] - now < 60 @staticmethod - def _is_scope_subset(needle_scope, haystack_scope): + def _is_scope_subset( + needle_scope: Optional[str], haystack_scope: Optional[str] + ) -> bool: needle_scope = set(needle_scope.split()) if needle_scope else set() - haystack_scope = ( - set(haystack_scope.split()) if haystack_scope else set() - ) + haystack_scope = set(haystack_scope.split()) if haystack_scope else set() return needle_scope <= haystack_scope - def _handle_oauth_error(self, http_error): + def _handle_oauth_error(self, http_error: requests.exceptions.HTTPError): response = http_error.response try: error_payload = response.json() @@ -133,12 +136,12 @@ class SpotifyClientCredentials(SpotifyAuthBase): def __init__( self, - client_id=None, - client_secret=None, - proxies=None, - requests_session=True, - requests_timeout=None, - cache_handler=None + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + proxies: Optional[Any] = None, + requests_session: Union[requests.Session, bool] = True, + requests_timeout: Optional[int] = None, + cache_handler: Optional[CacheHandler] = None, ): """ Creates a Client Credentials Flow Manager. @@ -181,7 +184,8 @@ class SpotifyClientCredentials(SpotifyAuthBase): else: self.cache_handler = CacheFileHandler() - def get_access_token(self, as_dict=True, check_cache=True): + # TODO: better return type based oninput type (overrides) + def get_access_token(self, as_dict: bool = True, check_cache: bool = True): """ If a valid access token is in memory, returns it Else fetches a new token and returns it @@ -254,20 +258,20 @@ class SpotifyOAuth(SpotifyAuthBase): OAUTH_TOKEN_URL = "https://accounts.spotify.com/api/token" def __init__( - self, - client_id=None, - client_secret=None, - redirect_uri=None, - state=None, - scope=None, - cache_path=None, - username=None, - proxies=None, - show_dialog=False, - requests_session=True, - requests_timeout=None, - open_browser=True, - cache_handler=None + self, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + redirect_uri: Optional[str] = None, + state: Optional[Any] = None, + scope: Optional[ScopeArgType] = None, + cache_path: Optional[str] = None, + username: Optional[str] = None, + proxies: Optional[Any] = None, + show_dialog: bool = False, + requests_session: Union[requests.Session, bool] = True, + requests_timeout: Optional[int] = None, + open_browser: bool = True, + cache_handler: Optional[CacheHandler] = None, ): """ Creates a SpotifyOAuth object @@ -352,7 +356,7 @@ class SpotifyOAuth(SpotifyAuthBase): return token_info - def get_authorize_url(self, state=None): + def get_authorize_url(self, state: Optional[Any] = None) -> str: """ Gets the URL to use to authorize this app """ payload = { @@ -405,7 +409,7 @@ class SpotifyOAuth(SpotifyAuthBase): except webbrowser.Error: logger.error(f"Please navigate here: {auth_url}") - def _get_auth_response_interactive(self, open_browser=False): + def _get_auth_response_interactive(self, open_browser: bool = False): if open_browser: self._open_auth_url() prompt = "Enter the URL you were redirected to: " @@ -435,7 +439,7 @@ class SpotifyOAuth(SpotifyAuthBase): else: raise SpotifyOauthError("Server listening on localhost has not been accessed") - def get_auth_response(self, open_browser=None): + def get_auth_response(self, open_browser: Optional[bool] = None): logger.info('User authentication requires interaction with your ' 'web browser. Once you enter your credentials and ' 'give authorization, you will be redirected to ' @@ -476,12 +480,14 @@ class SpotifyOAuth(SpotifyAuthBase): return self._get_auth_response_interactive(open_browser=open_browser) - def get_authorization_code(self, response=None): + def get_authorization_code(self, response: Optional[Any] = None): if response: return self.parse_response_code(response) return self.get_auth_response() - def get_access_token(self, code=None, as_dict=True, check_cache=True): + def get_access_token( + self, code: Optional[Any] = None, as_dict: bool = True, check_cache: bool = True + ): """ Gets the access token for the app given the code Parameters: @@ -540,7 +546,7 @@ class SpotifyOAuth(SpotifyAuthBase): except requests.exceptions.HTTPError as http_error: self._handle_oauth_error(http_error) - def refresh_access_token(self, refresh_token): + def refresh_access_token(self, refresh_token: str): payload = { "refresh_token": refresh_token, "grant_type": "refresh_token", @@ -619,18 +625,20 @@ class SpotifyPKCE(SpotifyAuthBase): OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize" OAUTH_TOKEN_URL = "https://accounts.spotify.com/api/token" - def __init__(self, - client_id=None, - redirect_uri=None, - state=None, - scope=None, - cache_path=None, - username=None, - proxies=None, - requests_timeout=None, - requests_session=True, - open_browser=True, - cache_handler=None): + def __init__( + self, + client_id: Optional[str] = None, + redirect_uri: Optional[str] = None, + state: Optional[Any] = None, + scope: Optional[ScopeArgType] = None, + cache_path: Optional[str] = None, + username: Optional[str] = None, + proxies: Optional[Any] = None, + requests_timeout: Optional[int] = None, + requests_session: Union[requests.Session, bool] = True, + open_browser: bool = True, + cache_handler: Optional[CacheHandler] = None, + ): """ Creates Auth Manager with the PKCE Auth flow. @@ -695,7 +703,7 @@ class SpotifyPKCE(SpotifyAuthBase): self.authorization_code = None self.open_browser = open_browser - def _get_code_verifier(self): + def _get_code_verifier(self) -> str: """ Spotify PCKE code verifier - See step 1 of the reference guide below Reference: https://developer.spotify.com/documentation/general/guides/authorization-guide/#authorization-code-flow-with-proof-key-for-code-exchange-pkce @@ -709,7 +717,7 @@ class SpotifyPKCE(SpotifyAuthBase): import secrets return secrets.token_urlsafe(length) - def _get_code_challenge(self): + def _get_code_challenge(self) -> str: """ Spotify PCKE code challenge - See step 1 of the reference guide below Reference: https://developer.spotify.com/documentation/general/guides/authorization-guide/#authorization-code-flow-with-proof-key-for-code-exchange-pkce @@ -720,7 +728,7 @@ class SpotifyPKCE(SpotifyAuthBase): code_challenge = base64.urlsafe_b64encode(code_challenge_digest).decode('utf-8') return code_challenge.replace('=', '') - def get_authorize_url(self, state=None): + def get_authorize_url(self, state: Optional[Any] = None) -> str: """ Gets the URL to use to authorize this app """ if not self.code_challenge: self.get_pkce_handshake_parameters() @@ -740,7 +748,7 @@ class SpotifyPKCE(SpotifyAuthBase): urlparams = urllibparse.urlencode(payload) return f"{self.OAUTH_AUTHORIZE_URL}?{urlparams}" - def _open_auth_url(self, state=None): + def _open_auth_url(self, state: Optional[Any] = None): auth_url = self.get_authorize_url(state) try: webbrowser.open(auth_url) @@ -748,7 +756,7 @@ class SpotifyPKCE(SpotifyAuthBase): except webbrowser.Error: logger.error(f"Please navigate here: {auth_url}") - def _get_auth_response(self, open_browser=None): + def _get_auth_response(self, open_browser: Optional[bool] = None): logger.info('User authentication requires interaction with your ' 'web browser. Once you enter your credentials and ' 'give authorization, you will be redirected to ' @@ -803,7 +811,7 @@ class SpotifyPKCE(SpotifyAuthBase): else: raise SpotifyOauthError("Server listening on localhost has not been accessed") - def _get_auth_response_interactive(self, open_browser=False): + def _get_auth_response_interactive(self, open_browser: bool = False): if open_browser or self.open_browser: self._open_auth_url() prompt = "Enter the URL you were redirected to: " @@ -817,7 +825,7 @@ class SpotifyPKCE(SpotifyAuthBase): raise SpotifyStateError(self.state, state) return code - def get_authorization_code(self, response=None): + def get_authorization_code(self, response: Optional[Any] = None): if response: return self.parse_response_code(response) return self._get_auth_response() @@ -851,7 +859,7 @@ class SpotifyPKCE(SpotifyAuthBase): self.code_verifier = self._get_code_verifier() self.code_challenge = self._get_code_challenge() - def get_access_token(self, code=None, check_cache=True): + def get_access_token(self, code: Optional[Any] = None, check_cache: bool = True): """ Gets the access token for the app If the code is not given and no cached token is used, an @@ -906,7 +914,7 @@ class SpotifyPKCE(SpotifyAuthBase): except requests.exceptions.HTTPError as http_error: self._handle_oauth_error(http_error) - def refresh_access_token(self, refresh_token): + def refresh_access_token(self, refresh_token: str): payload = { "refresh_token": refresh_token, "grant_type": "refresh_token", @@ -1008,15 +1016,17 @@ class SpotifyImplicitGrant(SpotifyAuthBase): """ OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize" - def __init__(self, - client_id=None, - redirect_uri=None, - state=None, - scope=None, - cache_path=None, - username=None, - show_dialog=False, - cache_handler=None): + def __init__( + self, + client_id: Optional[str] = None, + redirect_uri: Optional[str] = None, + state: Optional[Any] = None, + scope: Optional[ScopeArgType] = None, + cache_path: Optional[str] = None, + username: Optional[str] = None, + show_dialog: bool = False, + cache_handler: Optional[CacheHandler] = None, + ): """ Creates Auth Manager using the Implicit Grant flow **See help(SpotifyImplicitGrant) for full Security Warning** @@ -1092,10 +1102,12 @@ class SpotifyImplicitGrant(SpotifyAuthBase): return token_info - def get_access_token(self, - state=None, - response=None, - check_cache=True): + def get_access_token( + self, + state: Optional[Any] = None, + response: Optional[Any] = None, + check_cache: bool = True, + ): """ Gets Auth Token from cache (preferred) or user interaction Parameters @@ -1118,7 +1130,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase): return token_info["access_token"] - def get_authorize_url(self, state=None): + def get_authorize_url(self, state: Optional[Any] = None) -> str: """ Gets the URL to use to authorize this app """ payload = { "client_id": self.client_id, @@ -1138,7 +1150,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase): return f"{self.OAUTH_AUTHORIZE_URL}?{urlparams}" - def parse_response_token(self, url, state=None): + def parse_response_token(self, url, state: Optional[Any] = None): """ Parse the response code in the given response url """ remote_state, token, t_type, exp_in = self.parse_auth_response_url(url) if state is None: @@ -1163,7 +1175,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase): return tuple(form.get(param) for param in ["state", "access_token", "token_type", "expires_in"]) - def _open_auth_url(self, state=None): + def _open_auth_url(self, state: Optional[Any] = None): auth_url = self.get_authorize_url(state) try: webbrowser.open(auth_url) @@ -1171,7 +1183,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase): except webbrowser.Error: logger.error(f"Please navigate here: {auth_url}") - def get_auth_response(self, state=None): + def get_auth_response(self, state: Optional[Any] = None): """ Gets a new auth **token** with user interaction """ logger.info('User authentication requires interaction with your ' 'web browser. Once you enter your credentials and ' @@ -1274,7 +1286,7 @@ Close Window """) - def _write(self, text): + def _write(self, text: str): return self.wfile.write(text.encode("utf-8")) def log_message(self, format, *args): diff --git a/spotipy/util.py b/spotipy/util.py index bf9715f..1de82e3 100644 --- a/spotipy/util.py +++ b/spotipy/util.py @@ -8,6 +8,7 @@ import logging import os import warnings from types import TracebackType +from typing import Optional, Union, Tuple, List import requests import urllib3 @@ -26,17 +27,19 @@ CLIENT_CREDS_ENV_VARS = { # workaround for garbage collection REQUESTS_SESSION = requests.Session +StrListOrTuple = Union[List[str], Tuple[str, ...]] + def prompt_for_user_token( - username=None, - scope=None, - client_id=None, - client_secret=None, - redirect_uri=None, - cache_path=None, - oauth_manager=None, - show_dialog=False -): + username: Optional[str] = None, + scope: Optional[str] = None, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + redirect_uri: Optional[str] = None, + cache_path: Optional[str] = None, + oauth_manager: Optional[spotipy.SpotifyOAuth] = None, + show_dialog: bool = False, +) -> Union[str, None]: """ Prompt the user to login if necessary and returns a user token suitable for use with the spotipy.Spotify constructor. @@ -116,7 +119,7 @@ def prompt_for_user_token( return None -def get_host_port(netloc): +def get_host_port(netloc: str): """ Split the network location string into host and port and returns a tuple where the host is a string and the the port is an integer. @@ -132,8 +135,9 @@ def get_host_port(netloc): return host, port +ScopeArgType = Union[str, StrListOrTuple] -def normalize_scope(scope): +def normalize_scope(scope: Optional[ScopeArgType]) -> Union[str, None]: """Normalize the scope to verify that it is a list or tuple. A string input will split the string by commas to create a list of scopes. A list or tuple input is used directly. @@ -163,13 +167,13 @@ class Retry(urllib3.Retry): """ def increment( - self, - method: str | None = None, - url: str | None = None, - response: urllib3.BaseHTTPResponse | None = None, - error: Exception | None = None, - _pool: urllib3.connectionpool.ConnectionPool | None = None, - _stacktrace: TracebackType | None = None, + self, + method: Optional[str] = None, + url: Optional[str] = None, + response: Optional[urllib3.BaseHTTPResponse] = None, + error: Optional[Exception] = None, + _pool: Optional[urllib3.connectionpool.ConnectionPool] = None, + _stacktrace: Optional[TracebackType] = None, ) -> urllib3.Retry: if response: retry_header = response.headers.get("Retry-After") diff --git a/tests/helpers.py b/tests/helpers.py index 39321f0..1b46c93 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -1,9 +1,12 @@ import base64 +from typing import Union import requests +from spotipy import Spotify -def get_spotify_playlist(spotify_object, playlist_name, username): + +def get_spotify_playlist(spotify_object: Spotify, playlist_name: str, username: str): playlists = spotify_object.user_playlists(username) while playlists: for item in playlists['items']: @@ -12,5 +15,5 @@ def get_spotify_playlist(spotify_object, playlist_name, username): playlists = spotify_object.next(playlists) -def get_as_base64(url): +def get_as_base64(url: Union[str, bytes]) -> str: return base64.b64encode(requests.get(url).content).decode("utf-8") diff --git a/tests/unit/test_oauth.py b/tests/unit/test_oauth.py index 84bb0b4..53faeae 100644 --- a/tests/unit/test_oauth.py +++ b/tests/unit/test_oauth.py @@ -13,7 +13,7 @@ patch = mock.patch DEFAULT = mock.DEFAULT -def _make_fake_token(expires_at, expires_in, scope): +def _make_fake_token(expires_at: int, expires_in: int, scope: str): return dict( expires_at=expires_at, expires_in=expires_in,