This commit is contained in:
Blake V. 2026-02-16 20:30:38 +01:00 committed by GitHub
commit 56e494cb82
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 491 additions and 329 deletions

View File

@ -10,7 +10,7 @@ jobs:
SPOTIPY_CLIENT_SECRET: ${{ secrets.SPOTIPY_CLIENT_SECRET }} SPOTIPY_CLIENT_SECRET: ${{ secrets.SPOTIPY_CLIENT_SECRET }}
strategy: strategy:
matrix: matrix:
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }} - name: Set up Python ${{ matrix.python-version }}

View File

@ -7,7 +7,7 @@ jobs:
runs-on: ubuntu-22.04 runs-on: ubuntu-22.04
strategy: strategy:
matrix: matrix:
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }} - name: Set up Python ${{ matrix.python-version }}

View File

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

View File

@ -31,7 +31,7 @@ setup(
project_urls={ project_urls={
'Source': 'https://github.com/plamere/spotipy', 'Source': 'https://github.com/plamere/spotipy',
}, },
python_requires='>3.8', python_requires='>=3.10',
install_requires=[ install_requires=[
"redis>=3.5.3", # TODO: Move to extras_require in v3 "redis>=3.5.3", # TODO: Move to extras_require in v3
"requests>=2.25.0", "requests>=2.25.0",

View File

@ -11,6 +11,8 @@ import errno
import json import json
import logging import logging
import os import os
from json import JSONEncoder
from typing import Dict
from redis import RedisError from redis import RedisError
@ -36,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.
""" """
@ -49,10 +51,12 @@ class CacheFileHandler(CacheHandler):
as json files on disk. as json files on disk.
""" """
def __init__(self, def __init__(
cache_path=None, self,
username=None, cache_path: str | None = None,
encoder_cls=None): username: str | None = None,
encoder_cls: JSONEncoder | None = None,
):
""" """
Parameters: Parameters:
* cache_path: May be supplied, will otherwise be generated * cache_path: May be supplied, will otherwise be generated
@ -90,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))
@ -109,7 +113,7 @@ class MemoryCacheHandler(CacheHandler):
instance is freed. instance is freed.
""" """
def __init__(self, token_info=None): def __init__(self, token_info: Dict | None = 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.
@ -119,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
@ -148,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:
@ -173,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:
@ -185,7 +189,7 @@ class RedisCacheHandler(CacheHandler):
A cache handler that stores the token info in the Redis. A cache handler that stores the token info in the Redis.
""" """
def __init__(self, redis, key=None): def __init__(self, redis, key: str | None = None):
""" """
Parameters: Parameters:
* redis: Redis object provided by redis-py library * redis: Redis object provided by redis-py library
@ -194,7 +198,7 @@ class RedisCacheHandler(CacheHandler):
(takes precedence over `token_info`) (takes precedence over `token_info`)
""" """
self.redis = redis 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): def get_cached_token(self):
token_info = None token_info = None
@ -207,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:
@ -218,7 +222,7 @@ class MemcacheCacheHandler(CacheHandler):
"""A Cache handler that stores the token info in Memcache using the pymemcache client """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: str | None = None):
""" """
Parameters: Parameters:
* memcache: memcache client object provided by pymemcache * memcache: memcache client object provided by pymemcache
@ -227,7 +231,7 @@ class MemcacheCacheHandler(CacheHandler):
(takes precedence over `token_info`) (takes precedence over `token_info`)
""" """
self.memcache = memcache 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): def get_cached_token(self):
from pymemcache import MemcacheError from pymemcache import MemcacheError
@ -238,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))

File diff suppressed because it is too large Load Diff

View File

@ -34,8 +34,16 @@ class SpotifyOauthError(SpotifyBaseException):
class SpotifyStateError(SpotifyOauthError): class SpotifyStateError(SpotifyOauthError):
""" The state sent and state received were different """ """ The state sent and state received were different """
def __init__(self, local_state=None, remote_state=None, message=None, def __init__(
error=None, error_description=None, *args, **kwargs): self,
local_state=None,
remote_state=None,
message=None,
error=None,
error_description=None,
*args,
**kwargs,
):
if not message: if not message:
message = ("Expected " + local_state + " but received " message = ("Expected " + local_state + " but received "
+ remote_state) + remote_state)

View File

@ -16,6 +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, Dict, Union
from urllib.parse import parse_qsl, urlparse from urllib.parse import parse_qsl, urlparse
import requests import requests
@ -23,19 +24,21 @@ import requests
from spotipy.cache_handler import CacheFileHandler, CacheHandler from spotipy.cache_handler import CacheFileHandler, CacheHandler
from spotipy.exceptions import SpotifyOauthError, SpotifyStateError from spotipy.exceptions import SpotifyOauthError, SpotifyStateError
from spotipy.util import (CLIENT_CREDS_ENV_VARS, REQUESTS_SESSION, from spotipy.util import (CLIENT_CREDS_ENV_VARS, REQUESTS_SESSION,
get_host_port, normalize_scope) ScopeArgType, get_host_port, normalize_scope)
logger = logging.getLogger(__name__) 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( auth_header = base64.b64encode(
str(client_id + ":" + client_secret).encode("ascii") str(client_id + ":" + client_secret).encode("ascii")
) )
return {"Authorization": f"Basic {auth_header.decode('ascii')}"} return {"Authorization": f"Basic {auth_header.decode('ascii')}"}
def _ensure_value(value, env_key): def _ensure_value(value: str | None, env_key: str) -> str:
env_val = CLIENT_CREDS_ENV_VARS[env_key] env_val = CLIENT_CREDS_ENV_VARS[env_key]
_val = value or os.getenv(env_val) _val = value or os.getenv(env_val)
if _val is None: if _val is None:
@ -45,7 +48,7 @@ def _ensure_value(value, env_key):
class SpotifyAuthBase: class SpotifyAuthBase:
def __init__(self, requests_session): def __init__(self, requests_session: Union[requests.Session, bool] | None = None):
if isinstance(requests_session, requests.Session): if isinstance(requests_session, requests.Session):
self._session = requests_session self._session = requests_session
else: else:
@ -55,54 +58,54 @@ class SpotifyAuthBase:
from requests import api from requests import api
self._session = api self._session = api
def _normalize_scope(self, scope): def _normalize_scope(self, scope: ScopeArgType | None):
return normalize_scope(scope) return normalize_scope(scope)
@property @property
def client_id(self): def client_id(self) -> str:
return self._client_id return self._client_id
@client_id.setter @client_id.setter
def client_id(self, val): def client_id(self, val: str | None):
self._client_id = _ensure_value(val, "client_id") self._client_id = _ensure_value(val, "client_id")
@property @property
def client_secret(self): def client_secret(self) -> str:
return self._client_secret return self._client_secret
@client_secret.setter @client_secret.setter
def client_secret(self, val): def client_secret(self, val: str | None):
self._client_secret = _ensure_value(val, "client_secret") self._client_secret = _ensure_value(val, "client_secret")
@property @property
def redirect_uri(self): def redirect_uri(self) -> str:
return self._redirect_uri return self._redirect_uri
@redirect_uri.setter @redirect_uri.setter
def redirect_uri(self, val): def redirect_uri(self, val: str | None):
self._redirect_uri = _ensure_value(val, "redirect_uri") self._redirect_uri = _ensure_value(val, "redirect_uri")
@staticmethod @staticmethod
def _get_user_input(prompt): 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): 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
@staticmethod @staticmethod
def _is_scope_subset(needle_scope, haystack_scope): def _is_scope_subset(
needle_scope: str | None, haystack_scope: str | None
) -> bool:
needle_scope = set(needle_scope.split()) if needle_scope else set() needle_scope = set(needle_scope.split()) if needle_scope else set()
haystack_scope = ( haystack_scope = set(haystack_scope.split()) if haystack_scope else set()
set(haystack_scope.split()) if haystack_scope else set()
)
return needle_scope <= haystack_scope 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 response = http_error.response
try: try:
error_payload = response.json() error_payload = response.json()
@ -133,12 +136,12 @@ class SpotifyClientCredentials(SpotifyAuthBase):
def __init__( def __init__(
self, self,
client_id=None, client_id: str | None = None,
client_secret=None, client_secret: str | None = None,
proxies=None, proxies: Any | None = None,
requests_session=True, requests_session: Union[requests.Session, bool] = True,
requests_timeout=None, requests_timeout: int | None = None,
cache_handler=None cache_handler: CacheHandler | None = None,
): ):
""" """
Creates a Client Credentials Flow Manager. Creates a Client Credentials Flow Manager.
@ -181,7 +184,8 @@ class SpotifyClientCredentials(SpotifyAuthBase):
else: else:
self.cache_handler = CacheFileHandler() 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 If a valid access token is in memory, returns it
Else fetches a new token and returns it Else fetches a new token and returns it
@ -237,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.
@ -255,19 +259,19 @@ class SpotifyOAuth(SpotifyAuthBase):
def __init__( def __init__(
self, self,
client_id=None, client_id: str | None = None,
client_secret=None, client_secret: str | None = None,
redirect_uri=None, redirect_uri: str | None = None,
state=None, state: Any | None = None,
scope=None, scope: ScopeArgType | None = None,
cache_path=None, cache_path: str | None = None,
username=None, username: str | None = None,
proxies=None, proxies: Any | None = None,
show_dialog=False, show_dialog: bool = False,
requests_session=True, requests_session: Union[requests.Session, bool] = True,
requests_timeout=None, requests_timeout: int | None = None,
open_browser=True, open_browser: bool = True,
cache_handler=None cache_handler: CacheHandler | None = None,
): ):
""" """
Creates a SpotifyOAuth object Creates a SpotifyOAuth object
@ -335,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: Dict | None):
if token_info is None: if token_info is None:
return None return None
@ -352,7 +356,7 @@ class SpotifyOAuth(SpotifyAuthBase):
return token_info return token_info
def get_authorize_url(self, state=None): def get_authorize_url(self, state: Any | None = None) -> str:
""" Gets the URL to use to authorize this app """ Gets the URL to use to authorize this app
""" """
payload = { payload = {
@ -405,7 +409,7 @@ class SpotifyOAuth(SpotifyAuthBase):
except webbrowser.Error: except webbrowser.Error:
logger.error(f"Please navigate here: {auth_url}") 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: if open_browser:
self._open_auth_url() self._open_auth_url()
prompt = "Enter the URL you were redirected to: " prompt = "Enter the URL you were redirected to: "
@ -421,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()
@ -435,7 +439,7 @@ class SpotifyOAuth(SpotifyAuthBase):
else: else:
raise SpotifyOauthError("Server listening on localhost has not been accessed") 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: bool | None = None):
logger.info('User authentication requires interaction with your ' logger.info('User authentication requires interaction with your '
'web browser. Once you enter your credentials and ' 'web browser. Once you enter your credentials and '
'give authorization, you will be redirected to ' 'give authorization, you will be redirected to '
@ -476,12 +480,14 @@ class SpotifyOAuth(SpotifyAuthBase):
return self._get_auth_response_interactive(open_browser=open_browser) return self._get_auth_response_interactive(open_browser=open_browser)
def get_authorization_code(self, response=None): def get_authorization_code(self, response: Any | None = None):
if response: if response:
return self.parse_response_code(response) return self.parse_response_code(response)
return self.get_auth_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: str | None = 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
Parameters: Parameters:
@ -540,7 +546,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 refresh_access_token(self, refresh_token): def refresh_access_token(self, refresh_token: str):
payload = { payload = {
"refresh_token": refresh_token, "refresh_token": refresh_token,
"grant_type": "refresh_token", "grant_type": "refresh_token",
@ -569,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.
@ -593,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 " +
@ -619,18 +625,20 @@ class SpotifyPKCE(SpotifyAuthBase):
OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize" OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
OAUTH_TOKEN_URL = "https://accounts.spotify.com/api/token" OAUTH_TOKEN_URL = "https://accounts.spotify.com/api/token"
def __init__(self, def __init__(
client_id=None, self,
redirect_uri=None, client_id: str | None = None,
state=None, redirect_uri: str | None = None,
scope=None, state: Any | None = None,
cache_path=None, scope: ScopeArgType | None = None,
username=None, cache_path: str | None = None,
proxies=None, username: str | None = None,
requests_timeout=None, proxies: Any | None = None,
requests_session=True, requests_timeout: int | None = None,
open_browser=True, requests_session: Union[requests.Session, bool] = True,
cache_handler=None): open_browser: bool = True,
cache_handler: CacheHandler | None = None,
):
""" """
Creates Auth Manager with the PKCE Auth flow. Creates Auth Manager with the PKCE Auth flow.
@ -695,7 +703,7 @@ class SpotifyPKCE(SpotifyAuthBase):
self.authorization_code = None self.authorization_code = None
self.open_browser = open_browser 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 """ Spotify PCKE code verifier - See step 1 of the reference guide below
Reference: Reference:
https://developer.spotify.com/documentation/general/guides/authorization-guide/#authorization-code-flow-with-proof-key-for-code-exchange-pkce 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 import secrets
return secrets.token_urlsafe(length) 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 """ Spotify PCKE code challenge - See step 1 of the reference guide below
Reference: Reference:
https://developer.spotify.com/documentation/general/guides/authorization-guide/#authorization-code-flow-with-proof-key-for-code-exchange-pkce 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') code_challenge = base64.urlsafe_b64encode(code_challenge_digest).decode('utf-8')
return code_challenge.replace('=', '') return code_challenge.replace('=', '')
def get_authorize_url(self, state=None): def get_authorize_url(self, state: Any | None = None) -> str:
""" Gets the URL to use to authorize this app """ """ Gets the URL to use to authorize this app """
if not self.code_challenge: if not self.code_challenge:
self.get_pkce_handshake_parameters() self.get_pkce_handshake_parameters()
@ -740,7 +748,7 @@ class SpotifyPKCE(SpotifyAuthBase):
urlparams = urllibparse.urlencode(payload) urlparams = urllibparse.urlencode(payload)
return f"{self.OAUTH_AUTHORIZE_URL}?{urlparams}" return f"{self.OAUTH_AUTHORIZE_URL}?{urlparams}"
def _open_auth_url(self, state=None): def _open_auth_url(self, state: Any | None = None):
auth_url = self.get_authorize_url(state) auth_url = self.get_authorize_url(state)
try: try:
webbrowser.open(auth_url) webbrowser.open(auth_url)
@ -748,7 +756,7 @@ class SpotifyPKCE(SpotifyAuthBase):
except webbrowser.Error: except webbrowser.Error:
logger.error(f"Please navigate here: {auth_url}") logger.error(f"Please navigate here: {auth_url}")
def _get_auth_response(self, open_browser=None): def _get_auth_response(self, open_browser: bool | None = None):
logger.info('User authentication requires interaction with your ' logger.info('User authentication requires interaction with your '
'web browser. Once you enter your credentials and ' 'web browser. Once you enter your credentials and '
'give authorization, you will be redirected to ' 'give authorization, you will be redirected to '
@ -788,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()
@ -803,7 +811,7 @@ class SpotifyPKCE(SpotifyAuthBase):
else: else:
raise SpotifyOauthError("Server listening on localhost has not been accessed") 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: if open_browser or self.open_browser:
self._open_auth_url() self._open_auth_url()
prompt = "Enter the URL you were redirected to: " prompt = "Enter the URL you were redirected to: "
@ -817,7 +825,7 @@ class SpotifyPKCE(SpotifyAuthBase):
raise SpotifyStateError(self.state, state) raise SpotifyStateError(self.state, state)
return code return code
def get_authorization_code(self, response=None): def get_authorization_code(self, response: Any | None = None):
if response: if response:
return self.parse_response_code(response) return self.parse_response_code(response)
return self._get_auth_response() return self._get_auth_response()
@ -851,7 +859,7 @@ class SpotifyPKCE(SpotifyAuthBase):
self.code_verifier = self._get_code_verifier() self.code_verifier = self._get_code_verifier()
self.code_challenge = self._get_code_challenge() self.code_challenge = self._get_code_challenge()
def get_access_token(self, code=None, check_cache=True): def get_access_token(self, code: Any | None = None, check_cache: bool = True):
""" Gets the access token for the app """ Gets the access token for the app
If the code is not given and no cached token is used, an 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: except requests.exceptions.HTTPError as http_error:
self._handle_oauth_error(http_error) self._handle_oauth_error(http_error)
def refresh_access_token(self, refresh_token): def refresh_access_token(self, refresh_token: str):
payload = { payload = {
"refresh_token": refresh_token, "refresh_token": refresh_token,
"grant_type": "refresh_token", "grant_type": "refresh_token",
@ -962,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 " +
@ -1008,15 +1016,17 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
""" """
OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize" OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
def __init__(self, def __init__(
client_id=None, self,
redirect_uri=None, client_id: str | None = None,
state=None, redirect_uri: str | None = None,
scope=None, state: Any | None = None,
cache_path=None, scope: ScopeArgType | None = None,
username=None, cache_path: str | None = None,
show_dialog=False, username: str | None = None,
cache_handler=None): show_dialog: bool = False,
cache_handler: CacheHandler | None = None,
):
""" Creates Auth Manager using the Implicit Grant flow """ Creates Auth Manager using the Implicit Grant flow
**See help(SpotifyImplicitGrant) for full Security Warning** **See help(SpotifyImplicitGrant) for full Security Warning**
@ -1077,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: Dict | None):
if token_info is None: if token_info is None:
return None return None
@ -1092,10 +1102,12 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
return token_info return token_info
def get_access_token(self, def get_access_token(
state=None, self,
response=None, state: Any | None = None,
check_cache=True): response: Any | None = None,
check_cache: bool = True,
):
""" Gets Auth Token from cache (preferred) or user interaction """ Gets Auth Token from cache (preferred) or user interaction
Parameters Parameters
@ -1118,7 +1130,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
return token_info["access_token"] return token_info["access_token"]
def get_authorize_url(self, state=None): def get_authorize_url(self, state: Any | None = None) -> str:
""" Gets the URL to use to authorize this app """ """ Gets the URL to use to authorize this app """
payload = { payload = {
"client_id": self.client_id, "client_id": self.client_id,
@ -1138,7 +1150,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
return f"{self.OAUTH_AUTHORIZE_URL}?{urlparams}" return f"{self.OAUTH_AUTHORIZE_URL}?{urlparams}"
def parse_response_token(self, url, state=None): def parse_response_token(self, url, state: Any | None = None):
""" Parse the response code in the given response url """ """ Parse the response code in the given response url """
remote_state, token, t_type, exp_in = self.parse_auth_response_url(url) remote_state, token, t_type, exp_in = self.parse_auth_response_url(url)
if state is None: if state is None:
@ -1163,7 +1175,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
return tuple(form.get(param) for param in ["state", "access_token", return tuple(form.get(param) for param in ["state", "access_token",
"token_type", "expires_in"]) "token_type", "expires_in"])
def _open_auth_url(self, state=None): def _open_auth_url(self, state: Any | None = None):
auth_url = self.get_authorize_url(state) auth_url = self.get_authorize_url(state)
try: try:
webbrowser.open(auth_url) webbrowser.open(auth_url)
@ -1171,7 +1183,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
except webbrowser.Error: except webbrowser.Error:
logger.error(f"Please navigate here: {auth_url}") logger.error(f"Please navigate here: {auth_url}")
def get_auth_response(self, state=None): def get_auth_response(self, state: Any | None = None):
""" Gets a new auth **token** with user interaction """ """ Gets a new auth **token** with user interaction """
logger.info('User authentication requires interaction with your ' logger.info('User authentication requires interaction with your '
'web browser. Once you enter your credentials and ' 'web browser. Once you enter your credentials and '
@ -1200,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.
@ -1225,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 " +
@ -1274,14 +1286,14 @@ Close Window
</body> </body>
</html>""") </html>""")
def _write(self, text): def _write(self, text: str):
return self.wfile.write(text.encode("utf-8")) return self.wfile.write(text.encode("utf-8"))
def log_message(self, format, *args): def log_message(self, format, *args):
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

View File

@ -8,6 +8,7 @@ import logging
import os import os
import warnings import warnings
from types import TracebackType from types import TracebackType
from typing import List, Tuple, Union
import requests import requests
import urllib3 import urllib3
@ -26,17 +27,19 @@ CLIENT_CREDS_ENV_VARS = {
# workaround for garbage collection # workaround for garbage collection
REQUESTS_SESSION = requests.Session REQUESTS_SESSION = requests.Session
StrListOrTuple = Union[List[str], Tuple[str, ...]]
def prompt_for_user_token( def prompt_for_user_token(
username=None, username: str | None = None,
scope=None, scope: str | None = None,
client_id=None, client_id: str | None = None,
client_secret=None, client_secret: str | None = None,
redirect_uri=None, redirect_uri: str | None = None,
cache_path=None, cache_path: str | None = None,
oauth_manager=None, oauth_manager: spotipy.SpotifyOAuth | None = None,
show_dialog=False show_dialog: bool = False,
): ) -> str | None:
""" Prompt the user to login if necessary and returns a user token """ Prompt the user to login if necessary and returns a user token
suitable for use with the spotipy.Spotify constructor. suitable for use with the spotipy.Spotify constructor.
@ -116,7 +119,7 @@ def prompt_for_user_token(
return None 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 """ 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. where the host is a string and the the port is an integer.
@ -133,7 +136,10 @@ def get_host_port(netloc):
return host, port return host, port
def normalize_scope(scope): ScopeArgType = Union[str, StrListOrTuple]
def normalize_scope(scope: ScopeArgType | None) -> str | None:
"""Normalize the scope to verify that it is a list or tuple. A string """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. input will split the string by commas to create a list of scopes.
A list or tuple input is used directly. A list or tuple input is used directly.

View File

@ -1,9 +1,12 @@
import base64 import base64
from typing import Union
import requests 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) playlists = spotify_object.user_playlists(username)
while playlists: while playlists:
for item in playlists['items']: for item in playlists['items']:
@ -12,5 +15,5 @@ def get_spotify_playlist(spotify_object, playlist_name, username):
playlists = spotify_object.next(playlists) 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") return base64.b64encode(requests.get(url).content).decode("utf-8")

View File

@ -13,7 +13,7 @@ patch = mock.patch
DEFAULT = mock.DEFAULT 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( return dict(
expires_at=expires_at, expires_at=expires_at,
expires_in=expires_in, expires_in=expires_in,