A light weight Python library for the Spotify Web API
Token caching strategies and utility functions for efficient session management and scope handling. Spotipy provides multiple cache handler implementations for different environments and use cases.
Abstract base class defining the cache handler interface.
class CacheHandler:
"""
Abstract base class for token caching implementations.
All cache handlers must implement get_cached_token and save_token_to_cache methods.
"""
def get_cached_token(self):
"""
Get cached token information.
Returns:
dict: Token information dictionary or None if no valid token cached
"""
def save_token_to_cache(self, token_info):
"""
Save token information to cache.
Args:
token_info (dict): Token information dictionary to cache
Returns:
None
"""Store tokens in local files for persistent authentication across sessions.
class CacheFileHandler(CacheHandler):
def __init__(self, cache_path=None, username=None, encoder_cls=None):
"""
File-based token caching.
Args:
cache_path (str, optional): Path to cache file (default: .cache-{username})
username (str, optional): Username for cache file naming
encoder_cls (class, optional): JSON encoder class for serialization
"""
def get_cached_token(self):
"""
Get cached token from file.
Returns:
dict: Token information or None if file doesn't exist or invalid
"""
def save_token_to_cache(self, token_info):
"""
Save token to file.
Args:
token_info (dict): Token information to save
"""Store tokens in memory for the duration of the application session.
class MemoryCacheHandler(CacheHandler):
def __init__(self, token_info=None):
"""
In-memory token caching.
Args:
token_info (dict, optional): Initial token information
"""
def get_cached_token(self):
"""
Get cached token from memory.
Returns:
dict: Token information or None if not cached
"""
def save_token_to_cache(self, token_info):
"""
Save token to memory.
Args:
token_info (dict): Token information to save
"""Integration with popular Python web frameworks for session-based token storage.
class DjangoSessionCacheHandler(CacheHandler):
def __init__(self, request):
"""
Django session-based token caching.
Args:
request: Django HttpRequest object with session
"""
def get_cached_token(self):
"""Get cached token from Django session."""
def save_token_to_cache(self, token_info):
"""Save token to Django session."""
class FlaskSessionCacheHandler(CacheHandler):
def __init__(self, session=None):
"""
Flask session-based token caching.
Args:
session: Flask session object (uses flask.session if not provided)
"""
def get_cached_token(self):
"""Get cached token from Flask session."""
def save_token_to_cache(self, token_info):
"""Save token to Flask session."""Integration with external storage systems for scalable token management.
class RedisCacheHandler(CacheHandler):
def __init__(self, redis_instance=None, key=None, encoder_cls=None):
"""
Redis-based token caching.
Args:
redis_instance: Redis client instance (creates default if None)
key (str, optional): Redis key for token storage (default: spotipy_token)
encoder_cls (class, optional): JSON encoder class for serialization
"""
def get_cached_token(self):
"""Get cached token from Redis."""
def save_token_to_cache(self, token_info):
"""Save token to Redis."""
class MemcacheCacheHandler(CacheHandler):
def __init__(self, memcache_instance=None, key=None, encoder_cls=None):
"""
Memcache-based token caching.
Args:
memcache_instance: Memcache client instance
key (str, optional): Memcache key for token storage (default: spotipy_token)
encoder_cls (class, optional): JSON encoder class for serialization
"""
def get_cached_token(self):
"""Get cached token from Memcache."""
def save_token_to_cache(self, token_info):
"""Save token to Memcache."""Helper functions for OAuth scope management and URL parsing.
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):
"""
Prompt user for token (deprecated - use SpotifyOAuth instead).
Args:
username (str, optional): Spotify username
scope (str, optional): Desired scope of the request
client_id (str, optional): Client ID of your app
client_secret (str, optional): Client secret of your app
redirect_uri (str, optional): Redirect URI of your app
cache_path (str, optional): Path to location to save tokens
oauth_manager: OAuth manager object
show_dialog (bool): Show login prompt always (default: False)
Returns:
str: Access token or None if authentication fails
"""
def normalize_scope(scope):
"""
Normalize scope to verify that it is a list or tuple.
Args:
scope (str, list, tuple): Scope string or list/tuple of scopes
Returns:
str: Space-separated scope string or None
"""
def get_host_port(netloc):
"""
Split network location string into host and port.
Args:
netloc (str): Network location string
Returns:
tuple: (host, port) where host is string and port is int or None
"""Enhanced retry logic with rate limit warnings.
class Retry(urllib3.Retry):
"""
Custom retry class with rate limit warnings.
Extends urllib3.Retry to provide user-friendly warnings when rate limits are hit.
"""
def increment(self, method=None, url=None, response=None, error=None,
_pool=None, _stacktrace=None):
"""
Handle retry logic with enhanced rate limit messaging.
Args:
method (str, optional): HTTP method
url (str, optional): Request URL
response (urllib3.BaseHTTPResponse, optional): HTTP response
error (Exception, optional): Request error
_pool (urllib3.connectionpool.ConnectionPool, optional): Connection pool
_stacktrace (TracebackType, optional): Stack trace
Returns:
urllib3.Retry: Updated retry object
"""import spotipy
from spotipy.oauth2 import SpotifyOAuth
from spotipy.cache_handler import CacheFileHandler
# Custom cache file location
cache_handler = CacheFileHandler(cache_path=".spotify_cache", username="myuser")
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8080/callback",
scope="user-library-read",
cache_handler=cache_handler
)
sp = spotipy.Spotify(auth_manager=auth_manager)
# Token will be saved to .spotify_cache file
user = sp.current_user()
print(f"Authenticated as: {user['display_name']}")from spotipy.cache_handler import MemoryCacheHandler
# Memory-only caching (tokens lost when application exits)
memory_cache = MemoryCacheHandler()
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8080/callback",
scope="user-library-read",
cache_handler=memory_cache
)
# Manually set token if you have one
token_info = {
'access_token': 'your_access_token',
'token_type': 'Bearer',
'expires_in': 3600,
'refresh_token': 'your_refresh_token',
'scope': 'user-library-read',
'expires_at': 1640995200 # Unix timestamp
}
memory_cache.save_token_to_cache(token_info)
sp = spotipy.Spotify(auth_manager=auth_manager)import redis
from spotipy.cache_handler import RedisCacheHandler
# Redis setup
redis_client = redis.Redis(host='localhost', port=6379, db=0)
# Redis cache handler
redis_cache = RedisCacheHandler(
redis_instance=redis_client,
key="spotify_token_user123" # Unique key per user
)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8080/callback",
scope="user-library-read user-library-modify",
cache_handler=redis_cache
)
sp = spotipy.Spotify(auth_manager=auth_manager)
# Token will be stored in Redis with expiration
user = sp.current_user()
print(f"User: {user['display_name']}")
# Check Redis for stored token
stored_token = redis_client.get("spotify_token_user123")
if stored_token:
print("Token successfully stored in Redis")# views.py
from django.shortcuts import render, redirect
from spotipy.oauth2 import SpotifyOAuth
from spotipy.cache_handler import DjangoSessionCacheHandler
import spotipy
def spotify_login(request):
"""Handle Spotify OAuth login."""
cache_handler = DjangoSessionCacheHandler(request)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8000/spotify/callback/",
scope="user-library-read user-top-read",
cache_handler=cache_handler
)
# Get authorization URL
auth_url = auth_manager.get_authorization_url()
return redirect(auth_url)
def spotify_callback(request):
"""Handle Spotify OAuth callback."""
cache_handler = DjangoSessionCacheHandler(request)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8000/spotify/callback/",
scope="user-library-read user-top-read",
cache_handler=cache_handler
)
# Handle the callback
code = request.GET.get('code')
if code:
auth_manager.get_access_token(code)
return redirect('spotify_dashboard')
return redirect('spotify_login')
def spotify_dashboard(request):
"""Dashboard showing user's Spotify data."""
cache_handler = DjangoSessionCacheHandler(request)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8000/spotify/callback/",
scope="user-library-read user-top-read",
cache_handler=cache_handler
)
# Check if user is authenticated
token_info = cache_handler.get_cached_token()
if not token_info:
return redirect('spotify_login')
sp = spotipy.Spotify(auth_manager=auth_manager)
try:
user = sp.current_user()
top_tracks = sp.current_user_top_tracks(limit=10)
context = {
'user': user,
'top_tracks': top_tracks['items']
}
return render(request, 'spotify_dashboard.html', context)
except Exception as e:
# Token might be expired, redirect to login
return redirect('spotify_login')from flask import Flask, session, request, redirect, url_for, render_template
from spotipy.oauth2 import SpotifyOAuth
from spotipy.cache_handler import FlaskSessionCacheHandler
import spotipy
app = Flask(__name__)
app.secret_key = 'your-secret-key'
@app.route('/')
def index():
cache_handler = FlaskSessionCacheHandler(session)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri=url_for('spotify_callback', _external=True),
scope="user-library-read",
cache_handler=cache_handler
)
if request.args.get("code"):
# Handle callback
auth_manager.get_access_token(request.args.get("code"))
return redirect(url_for('dashboard'))
if not auth_manager.validate_token(cache_handler.get_cached_token()):
# Need authentication
auth_url = auth_manager.get_authorization_url()
return f'<a href="{auth_url}">Login with Spotify</a>'
return redirect(url_for('dashboard'))
@app.route('/callback')
def spotify_callback():
return redirect(url_for('index'))
@app.route('/dashboard')
def dashboard():
cache_handler = FlaskSessionCacheHandler(session)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri=url_for('spotify_callback', _external=True),
scope="user-library-read",
cache_handler=cache_handler
)
if not auth_manager.validate_token(cache_handler.get_cached_token()):
return redirect(url_for('index'))
sp = spotipy.Spotify(auth_manager=auth_manager)
user = sp.current_user()
return f"Hello {user['display_name']}! <a href='{url_for('logout')}'>Logout</a>"
@app.route('/logout')
def logout():
session.clear()
return redirect(url_for('index'))
if __name__ == '__main__':
app.run(debug=True)from spotipy.cache_handler import CacheHandler
import json
import os
from datetime import datetime
class DatabaseCacheHandler(CacheHandler):
"""Custom cache handler using database storage."""
def __init__(self, user_id, db_connection):
self.user_id = user_id
self.db = db_connection
def get_cached_token(self):
"""Get token from database."""
cursor = self.db.cursor()
cursor.execute(
"SELECT token_data FROM spotify_tokens WHERE user_id = %s",
(self.user_id,)
)
result = cursor.fetchone()
if result:
try:
token_info = json.loads(result[0])
# Check if token is expired
if token_info.get('expires_at', 0) > datetime.now().timestamp():
return token_info
except json.JSONDecodeError:
pass
return None
def save_token_to_cache(self, token_info):
"""Save token to database."""
cursor = self.db.cursor()
cursor.execute("""
INSERT INTO spotify_tokens (user_id, token_data, updated_at)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE
token_data = VALUES(token_data),
updated_at = VALUES(updated_at)
""", (
self.user_id,
json.dumps(token_info),
datetime.now()
))
self.db.commit()
# Usage with custom handler
# db_cache = DatabaseCacheHandler("user123", mysql_connection)
# auth_manager = SpotifyOAuth(cache_handler=db_cache)class MultiUserCacheManager:
"""Manage tokens for multiple users with different cache strategies."""
def __init__(self):
self.cache_handlers = {}
def get_user_cache(self, user_id, cache_type="file"):
"""Get cache handler for specific user."""
if user_id not in self.cache_handlers:
if cache_type == "file":
cache_handler = CacheFileHandler(
cache_path=f".cache-{user_id}",
username=user_id
)
elif cache_type == "memory":
cache_handler = MemoryCacheHandler()
elif cache_type == "redis":
redis_client = redis.Redis()
cache_handler = RedisCacheHandler(
redis_instance=redis_client,
key=f"spotify_token_{user_id}"
)
else:
raise ValueError(f"Unknown cache type: {cache_type}")
self.cache_handlers[user_id] = cache_handler
return self.cache_handlers[user_id]
def get_spotify_client(self, user_id, scope="user-library-read", cache_type="file"):
"""Get authenticated Spotify client for user."""
cache_handler = self.get_user_cache(user_id, cache_type)
auth_manager = SpotifyOAuth(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="http://localhost:8080/callback",
scope=scope,
cache_handler=cache_handler
)
return spotipy.Spotify(auth_manager=auth_manager)
def clear_user_cache(self, user_id):
"""Clear cached token for user."""
if user_id in self.cache_handlers:
# For file cache, remove the file
cache_handler = self.cache_handlers[user_id]
if isinstance(cache_handler, CacheFileHandler):
try:
os.remove(f".cache-{user_id}")
except FileNotFoundError:
pass
del self.cache_handlers[user_id]
# Usage
cache_manager = MultiUserCacheManager()
# Get client for different users
user1_sp = cache_manager.get_spotify_client("user1", cache_type="redis")
user2_sp = cache_manager.get_spotify_client("user2", cache_type="file")
# Use clients
user1_profile = user1_sp.current_user()
user2_profile = user2_sp.current_user()from spotipy.util import normalize_scope
# Normalize different scope formats
scopes = [
"user-library-read,user-library-modify", # Comma-separated string
["user-library-read", "user-library-modify"], # List
("user-library-read", "user-library-modify"), # Tuple
]
for scope in scopes:
normalized = normalize_scope(scope)
print(f"Input: {scope}")
print(f"Normalized: '{normalized}'")
print()
# Build scopes programmatically
def build_scope_for_features(features):
"""Build OAuth scope based on required features."""
scope_mapping = {
'library': ['user-library-read', 'user-library-modify'],
'playlists': ['playlist-read-private', 'playlist-modify-private', 'playlist-modify-public'],
'playback': ['user-read-playback-state', 'user-modify-playback-state'],
'following': ['user-follow-read', 'user-follow-modify'],
'top_content': ['user-top-read'],
'recently_played': ['user-read-recently-played'],
'profile': ['user-read-private', 'user-read-email']
}
required_scopes = []
for feature in features:
if feature in scope_mapping:
required_scopes.extend(scope_mapping[feature])
# Remove duplicates and normalize
unique_scopes = list(set(required_scopes))
return normalize_scope(unique_scopes)
# Example usage
app_features = ['library', 'playlists', 'playback']
required_scope = build_scope_for_features(app_features)
print(f"Required scope: {required_scope}")Configure cache and authentication settings using environment variables:
# Client credentials
export SPOTIPY_CLIENT_ID='your_client_id'
export SPOTIPY_CLIENT_SECRET='your_client_secret'
export SPOTIPY_REDIRECT_URI='http://localhost:8080/callback'
# Cache settings
export SPOTIPY_CACHE_PATH='.spotify_cache'
export SPOTIPY_CACHE_USERNAME='default_user'
# Redis settings (if using Redis cache)
export REDIS_URL='redis://localhost:6379/0'
export SPOTIPY_REDIS_KEY='spotify_tokens'Install with Tessl CLI
npx tessl i tessl/pypi-spotipy