token_info dict type

This commit is contained in:
bvandercar-vt 2025-12-02 16:07:03 -07:00
parent 0a797b5cad
commit 4328005af8
4 changed files with 44 additions and 34 deletions

View File

@ -11,7 +11,7 @@ Add your changes below.
### Added ### Added
- Adds types to function args - Adds type hints to all function args
### Fixed ### Fixed

View File

@ -12,7 +12,7 @@ import json
import logging import logging
import os import os
from json import JSONEncoder from json import JSONEncoder
from typing import Optional from typing import Dict, Optional
from redis import RedisError from redis import RedisError
@ -38,7 +38,7 @@ class CacheHandler():
# return token_info # return token_info
raise NotImplementedError() raise NotImplementedError()
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
""" """
Save a token_info dictionary object to the cache and return None. Save a token_info dictionary object to the cache and return None.
""" """
@ -94,7 +94,7 @@ class CacheFileHandler(CacheHandler):
return token_info return token_info
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
try: try:
with open(self.cache_path, "w", encoding='utf-8') as f: with open(self.cache_path, "w", encoding='utf-8') as f:
f.write(json.dumps(token_info, cls=self.encoder_cls)) f.write(json.dumps(token_info, cls=self.encoder_cls))
@ -113,7 +113,7 @@ class MemoryCacheHandler(CacheHandler):
instance is freed. instance is freed.
""" """
def __init__(self, token_info=None): def __init__(self, token_info: Optional[Dict] = None):
""" """
Parameters: Parameters:
* token_info: The token info to store in memory. Can be None. * token_info: The token info to store in memory. Can be None.
@ -123,7 +123,7 @@ class MemoryCacheHandler(CacheHandler):
def get_cached_token(self): def get_cached_token(self):
return self.token_info return self.token_info
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
self.token_info = token_info self.token_info = token_info
@ -152,7 +152,7 @@ class DjangoSessionCacheHandler(CacheHandler):
return token_info return token_info
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
try: try:
self.request.session['token_info'] = token_info self.request.session['token_info'] = token_info
except Exception as e: except Exception as e:
@ -177,7 +177,7 @@ class FlaskSessionCacheHandler(CacheHandler):
return token_info return token_info
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
try: try:
self.session["token_info"] = token_info self.session["token_info"] = token_info
except Exception as e: except Exception as e:
@ -211,7 +211,7 @@ class RedisCacheHandler(CacheHandler):
return token_info return token_info
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
try: try:
self.redis.set(self.key, json.dumps(token_info)) self.redis.set(self.key, json.dumps(token_info))
except RedisError as e: except RedisError as e:
@ -242,7 +242,7 @@ class MemcacheCacheHandler(CacheHandler):
except MemcacheError as e: except MemcacheError as e:
logger.warning(f"Error getting token to cache: {e}") logger.warning(f"Error getting token to cache: {e}")
def save_token_to_cache(self, token_info): def save_token_to_cache(self, token_info: Dict):
from pymemcache import MemcacheError from pymemcache import MemcacheError
try: try:
self.memcache.set(self.key, json.dumps(token_info)) self.memcache.set(self.key, json.dumps(token_info))

View File

@ -694,7 +694,7 @@ class Spotify:
def playlist( def playlist(
self, self,
playlist_id: str, playlist_id: str,
fields=None, fields: Optional[str] = None,
market: Optional[str] = None, market: Optional[str] = None,
additional_types: StrListOrTuple = ("track",), additional_types: StrListOrTuple = ("track",),
): ):
@ -719,7 +719,7 @@ class Spotify:
def playlist_tracks( def playlist_tracks(
self, self,
playlist_id: str, playlist_id: str,
fields=None, fields: Optional[str] = None,
limit: int = 100, limit: int = 100,
offset: int = 0, offset: int = 0,
market: Optional[str] = None, market: Optional[str] = None,
@ -750,8 +750,8 @@ class Spotify:
def playlist_items( def playlist_items(
self, self,
playlist_id, playlist_id: str,
fields=None, fields: Optional[str] = None,
limit: int = 100, limit: int = 100,
offset: int = 0, offset: int = 0,
market: Optional[str] = None, market: Optional[str] = None,
@ -1989,7 +1989,9 @@ class Spotify:
""" """
return self._get("me/player/devices") return self._get("me/player/devices")
def current_playback(self, market: Optional[str] = None, additional_types=None): def current_playback(
self, market: Optional[str] = None, additional_types: Optional[str] = None
):
""" Get information about user's current playback. """ Get information about user's current playback.
Parameters: Parameters:
@ -1998,7 +2000,9 @@ class Spotify:
""" """
return self._get("me/player", market=market, additional_types=additional_types) return self._get("me/player", market=market, additional_types=additional_types)
def currently_playing(self, market: Optional[str] = None, additional_types=None): def currently_playing(
self, market: Optional[str] = None, additional_types: Optional[str] = None
):
""" Get user's currently playing track. """ Get user's currently playing track.
Parameters: Parameters:
@ -2149,7 +2153,7 @@ class Spotify:
) )
) )
def shuffle(self, state, device_id: Optional[str] = None): def shuffle(self, state: bool, device_id: Optional[str] = None):
""" Toggle playback shuffling. """ Toggle playback shuffling.
Parameters: Parameters:
@ -2170,7 +2174,7 @@ class Spotify:
""" Gets the current user's queue """ """ Gets the current user's queue """
return self._get("me/player/queue") return self._get("me/player/queue")
def add_to_queue(self, uri, device_id: Optional[str] = None): def add_to_queue(self, uri: str, device_id: Optional[str] = None):
""" Adds a song to the end of a user's queue """ 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 If device A is currently playing music, and you try to add to the queue
@ -2248,7 +2252,13 @@ class Spotify:
return re.search(Spotify._regex_spotify_uri, uri) is not None return re.search(Spotify._regex_spotify_uri, uri) is not None
def _search_multiple_markets( def _search_multiple_markets(
self, q: str, limit: int, offset: int, type, markets: StrListOrTuple, total: int self,
q: str,
limit: int,
offset: int,
type: str,
markets: StrListOrTuple,
total: Optional[int],
): ):
if total and limit > total: if total and limit > total:
limit = total limit = total

View File

@ -16,7 +16,7 @@ import urllib.parse as urllibparse
import warnings import warnings
import webbrowser import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any, Optional, Union from typing import Any, Dict, Optional, Union
from urllib.parse import parse_qsl, urlparse from urllib.parse import parse_qsl, urlparse
import requests import requests
@ -86,14 +86,14 @@ class SpotifyAuthBase:
self._redirect_uri = _ensure_value(val, "redirect_uri") self._redirect_uri = _ensure_value(val, "redirect_uri")
@staticmethod @staticmethod
def _get_user_input(prompt) -> str: def _get_user_input(prompt: Union[str, object]) -> str:
try: try:
return raw_input(prompt) return raw_input(prompt)
except NameError: except NameError:
return input(prompt) return input(prompt)
@staticmethod @staticmethod
def is_token_expired(token_info) -> bool: def is_token_expired(token_info: Dict):
now = int(time.time()) now = int(time.time())
return token_info["expires_at"] - now < 60 return token_info["expires_at"] - now < 60
@ -241,7 +241,7 @@ class SpotifyClientCredentials(SpotifyAuthBase):
except requests.exceptions.HTTPError as http_error: except requests.exceptions.HTTPError as http_error:
self._handle_oauth_error(http_error) self._handle_oauth_error(http_error)
def _add_custom_values_to_token_info(self, token_info): def _add_custom_values_to_token_info(self, token_info: Dict):
""" """
Store some values that aren't directly provided by a Web API Store some values that aren't directly provided by a Web API
response. response.
@ -339,7 +339,7 @@ class SpotifyOAuth(SpotifyAuthBase):
self.show_dialog = show_dialog self.show_dialog = show_dialog
self.open_browser = open_browser self.open_browser = open_browser
def validate_token(self, token_info): def validate_token(self, token_info: Optional[Dict]):
if token_info is None: if token_info is None:
return None return None
@ -425,7 +425,7 @@ class SpotifyOAuth(SpotifyAuthBase):
raise SpotifyStateError(self.state, state) raise SpotifyStateError(self.state, state)
return code return code
def _get_auth_response_local_server(self, redirect_port): def _get_auth_response_local_server(self, redirect_port: int):
server = start_local_http_server(redirect_port) server = start_local_http_server(redirect_port)
self._open_auth_url() self._open_auth_url()
server.handle_request() server.handle_request()
@ -486,7 +486,7 @@ class SpotifyOAuth(SpotifyAuthBase):
return self.get_auth_response() return self.get_auth_response()
def get_access_token( def get_access_token(
self, code: Optional[Any] = None, as_dict: bool = True, check_cache: bool = True self, code: Optional[str] = None, as_dict: bool = True, check_cache: bool = True
): ):
""" Gets the access token for the app given the code """ Gets the access token for the app given the code
@ -575,7 +575,7 @@ class SpotifyOAuth(SpotifyAuthBase):
except requests.exceptions.HTTPError as http_error: except requests.exceptions.HTTPError as http_error:
self._handle_oauth_error(http_error) self._handle_oauth_error(http_error)
def _add_custom_values_to_token_info(self, token_info): def _add_custom_values_to_token_info(self, token_info: Dict):
""" """
Store some values that aren't directly provided by a Web API Store some values that aren't directly provided by a Web API
response. response.
@ -599,7 +599,7 @@ class SpotifyOAuth(SpotifyAuthBase):
) )
return self.validate_token(self.cache_handler.get_cached_token()) return self.validate_token(self.cache_handler.get_cached_token())
def _save_token_info(self, token_info): def _save_token_info(self, token_info: Dict):
warnings.warn("Calling _save_token_info directly on the SpotifyOAuth object will be " + warnings.warn("Calling _save_token_info directly on the SpotifyOAuth object will be " +
"deprecated. Instead, please specify a CacheFileHandler instance as " + "deprecated. Instead, please specify a CacheFileHandler instance as " +
"the cache_handler in SpotifyOAuth and use the CacheFileHandler's " + "the cache_handler in SpotifyOAuth and use the CacheFileHandler's " +
@ -796,7 +796,7 @@ class SpotifyPKCE(SpotifyAuthBase):
'the URL your browser is redirected to.') 'the URL your browser is redirected to.')
return self._get_auth_response_interactive(open_browser=open_browser) return self._get_auth_response_interactive(open_browser=open_browser)
def _get_auth_response_local_server(self, redirect_port): def _get_auth_response_local_server(self, redirect_port: int):
server = start_local_http_server(redirect_port) server = start_local_http_server(redirect_port)
self._open_auth_url() self._open_auth_url()
server.handle_request() server.handle_request()
@ -970,7 +970,7 @@ class SpotifyPKCE(SpotifyAuthBase):
) )
return self.validate_token(self.cache_handler.get_cached_token()) return self.validate_token(self.cache_handler.get_cached_token())
def _save_token_info(self, token_info): def _save_token_info(self, token_info: Dict):
warnings.warn("Calling _save_token_info directly on the SpotifyOAuth object will be " + warnings.warn("Calling _save_token_info directly on the SpotifyOAuth object will be " +
"deprecated. Instead, please specify a CacheFileHandler instance as " + "deprecated. Instead, please specify a CacheFileHandler instance as " +
"the cache_handler in SpotifyOAuth and use the CacheFileHandler's " + "the cache_handler in SpotifyOAuth and use the CacheFileHandler's " +
@ -1087,7 +1087,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
self.show_dialog = show_dialog self.show_dialog = show_dialog
self._session = None # As to not break inherited __del__ self._session = None # As to not break inherited __del__
def validate_token(self, token_info): def validate_token(self, token_info: Optional[Dict]):
if token_info is None: if token_info is None:
return None return None
@ -1212,7 +1212,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
"were redirected to: ") "were redirected to: ")
return self.parse_response_token(response, state) return self.parse_response_token(response, state)
def _add_custom_values_to_token_info(self, token_info): def _add_custom_values_to_token_info(self, token_info: Dict):
""" """
Store some values that aren't directly provided by a Web API Store some values that aren't directly provided by a Web API
response. response.
@ -1237,7 +1237,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
) )
return self.validate_token(self.cache_handler.get_cached_token()) return self.validate_token(self.cache_handler.get_cached_token())
def _save_token_info(self, token_info): def _save_token_info(self, token_info: Dict):
warnings.warn("Calling _save_token_info directly on the SpotifyImplicitGrant " + warnings.warn("Calling _save_token_info directly on the SpotifyImplicitGrant " +
"object will be deprecated. Instead, please specify a " + "object will be deprecated. Instead, please specify a " +
"CacheFileHandler instance as the cache_handler in SpotifyOAuth " + "CacheFileHandler instance as the cache_handler in SpotifyOAuth " +
@ -1293,7 +1293,7 @@ Close Window
return return
def start_local_http_server(port, handler=RequestHandler): def start_local_http_server(port: int, handler=RequestHandler):
server = HTTPServer(("127.0.0.1", port), handler) server = HTTPServer(("127.0.0.1", port), handler)
server.allow_reuse_address = True server.allow_reuse_address = True
server.auth_code = None server.auth_code = None