The API is primarily used by the Django app itself, and primarily probably as part of handling user requests in views. However, it may also be called asynchronously or externally in the future for integrations or background processing. This really is just a learning exercise for me to see how I can build a website with Django and integrate an API. I do not have many goals beyond “does it work?” and to get some of the basic aforementioned metadata on a page or two. I plan on establishing more goals as I progress with the project.
I’ve got a basic integration working. Here are some of the project’s files, if you could perhaps take a look:
core/spotify.py
import base64
import json
import secrets
import time
import urllib.error
import urllib.parse
import urllib.request
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
AUTH_BASE_URL = 'https://accounts.spotify.com/authorize'
TOKEN_URL = 'https://accounts.spotify.com/api/token'
API_BASE_URL = 'https://api.spotify.com/v1'
class SpotifyAPIError(Exception):
def __init__(self, status_code, payload):
self.status_code = status_code
self.payload = payload
super().__init__(f'Spotify API error ({status_code})')
# Internal helpers (module-private).
# Load Spotify settings and ensure required values exist.
def _get_config():
client_id = settings.SPOTIFY_CLIENT_ID
client_secret = settings.SPOTIFY_CLIENT_SECRET
redirect_uri = settings.SPOTIFY_REDIRECT_URI
scopes = settings.SPOTIFY_SCOPES
missing = []
if not client_id:
missing.append('SPOTIFY_CLIENT_ID')
if not client_secret:
missing.append('SPOTIFY_CLIENT_SECRET')
if not redirect_uri:
missing.append('SPOTIFY_REDIRECT_URI')
if missing:
missing_str = ', '.join(missing)
raise ImproperlyConfigured(
f'Missing Spotify settings: {missing_str}',
)
return {
'client_id': client_id,
'client_secret': client_secret,
'redirect_uri': redirect_uri,
'scopes': scopes or '',
}
# Create HTTP Basic auth header for token endpoint.
def _basic_auth_header(client_id, client_secret):
token_bytes = f'{client_id}:{client_secret}'.encode('utf-8')
encoded = base64.b64encode(token_bytes).decode('utf-8')
return f'Basic {encoded}'
# Decode JSON response body.
def _read_json(response):
payload = response.read().decode('utf-8')
if not payload:
return {}
return json.loads(payload)
# POST x-www-form-urlencoded data and parse JSON.
def _send_form(url, data, headers):
encoded = urllib.parse.urlencode(data).encode('utf-8')
request = urllib.request.Request(url, data=encoded, method='POST')
request.add_header('Content-Type', 'application/x-www-form-urlencoded')
for key, value in headers.items():
request.add_header(key, value)
try:
with urllib.request.urlopen(request) as response:
return _read_json(response)
except urllib.error.HTTPError as exc:
try:
payload = _read_json(exc)
except json.JSONDecodeError:
payload = {'error': exc.read().decode('utf-8')}
raise SpotifyAPIError(exc.code, payload) from exc
# Public API used by views.
# Build authorization URL and stash CSRF state in session.
def build_authorize_url(request):
config = _get_config()
state = secrets.token_urlsafe(16)
request.session['spotify_auth_state'] = state
params = {
'response_type': 'code',
'client_id': config['client_id'],
'redirect_uri': config['redirect_uri'],
'scope': config['scopes'],
'state': state,
}
return f'{AUTH_BASE_URL}?{urllib.parse.urlencode(params)}'
# Exchange auth code for access/refresh tokens.
def exchange_code_for_token(code):
config = _get_config()
headers = {
'Authorization': _basic_auth_header(
config['client_id'],
config['client_secret'],
),
}
data = {
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': config['redirect_uri'],
}
return _send_form(TOKEN_URL, data, headers)
# Refresh access token using refresh token.
def refresh_access_token(refresh_token):
config = _get_config()
headers = {
'Authorization': _basic_auth_header(
config['client_id'],
config['client_secret'],
),
}
data = {
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
}
return _send_form(TOKEN_URL, data, headers)
# Return cached access token or refresh if expired.
def get_valid_access_token(request):
access_token = request.session.get('spotify_access_token')
expires_at = request.session.get('spotify_expires_at')
refresh_token = request.session.get('spotify_refresh_token')
now = int(time.time())
if access_token and expires_at and now < int(expires_at):
return access_token
if access_token and not expires_at:
return access_token
if refresh_token:
token_data = refresh_access_token(refresh_token)
access_token = token_data.get('access_token')
expires_in = token_data.get('expires_in', 0)
if access_token:
request.session['spotify_access_token'] = access_token
new_refresh_token = token_data.get('refresh_token')
if new_refresh_token:
request.session['spotify_refresh_token'] = new_refresh_token
if expires_in:
request.session['spotify_expires_at'] = now + int(expires_in) - 30
return access_token
return None
# Call Spotify Web API with bearer token.
def spotify_api_get(path, access_token):
url = f'{API_BASE_URL}/{path.lstrip("/")}'
request = urllib.request.Request(url, method='GET')
request.add_header('Authorization', f'Bearer {access_token}')
try:
with urllib.request.urlopen(request) as response:
return _read_json(response)
except urllib.error.HTTPError as exc:
try:
payload = _read_json(exc)
except json.JSONDecodeError:
payload = {'error': exc.read().decode('utf-8')}
raise SpotifyAPIError(exc.code, payload) from exc
core/views.py
import time
from django.http import HttpResponse, JsonResponse
from django.shortcuts import redirect
from django.views.decorators.http import require_GET
from .spotify import (
SpotifyAPIError,
build_authorize_url,
exchange_code_for_token,
get_valid_access_token,
spotify_api_get,
)
# Simple health/landing endpoint.
@require_GET
def home(request):
return HttpResponse(
'Djotify is running. Visit /spotify/login to connect your Spotify account.',
)
# Redirect to Spotify's OAuth consent page.
@require_GET
def spotify_login(request):
return redirect(build_authorize_url(request))
# Handle OAuth redirect, verify state, store tokens.
@require_GET
def spotify_callback(request):
error = request.GET.get('error')
if error:
return JsonResponse({'error': error}, status=400)
state = request.GET.get('state')
if not state or state != request.session.get('spotify_auth_state'):
return JsonResponse({'error': 'Invalid state'}, status=400)
request.session.pop('spotify_auth_state', None)
code = request.GET.get('code')
if not code:
return JsonResponse({'error': 'Missing code'}, status=400)
try:
token_data = exchange_code_for_token(code)
except SpotifyAPIError as exc:
return JsonResponse(
{'error': 'Token exchange failed', 'details': exc.payload},
status=exc.status_code,
)
access_token = token_data.get('access_token')
refresh_token = token_data.get('refresh_token')
expires_in = token_data.get('expires_in', 0)
if access_token:
request.session['spotify_access_token'] = access_token
else:
return JsonResponse({'error': 'No access token returned'}, status=400)
if refresh_token:
request.session['spotify_refresh_token'] = refresh_token
if expires_in:
request.session['spotify_expires_at'] = int(time.time()) + int(expires_in) - 30
return redirect('spotify-me')
# Fetch current Spotify profile.
@require_GET
def spotify_me(request):
try:
access_token = get_valid_access_token(request)
except SpotifyAPIError as exc:
return JsonResponse(
{'error': 'Token refresh failed', 'details': exc.payload},
status=exc.status_code,
)
if not access_token:
return redirect('spotify-login')
try:
profile = spotify_api_get('me', access_token)
except SpotifyAPIError as exc:
return JsonResponse(
{'error': 'Spotify API request failed', 'details': exc.payload},
status=exc.status_code,
)
return JsonResponse(profile)
# Clear tokens and state from session.
@require_GET
def spotify_logout(request):
for key in [
'spotify_access_token',
'spotify_refresh_token',
'spotify_expires_at',
'spotify_auth_state',
]:
request.session.pop(key, None)
return redirect('home')
core/urls.py
from django.urls import path
from . import views
urlpatterns = [
path('', views.home, name='home'),
path('spotify/login/', views.spotify_login, name='spotify-login'),
path('spotify/callback/', views.spotify_callback, name='spotify-callback'),
path('spotify/me/', views.spotify_me, name='spotify-me'),
path('spotify/logout/', views.spotify_logout, name='spotify-logout'),
]
djotify/settings.py
# ...
# Spotify Web API settings
SPOTIFY_CLIENT_ID = os.environ.get('SPOTIFY_CLIENT_ID', '')
SPOTIFY_CLIENT_SECRET = os.environ.get('SPOTIFY_CLIENT_SECRET', '')
SPOTIFY_REDIRECT_URI = os.environ.get(
'SPOTIFY_REDIRECT_URI',
'http://127.0.0.1:8000/spotify/callback/',
)
SPOTIFY_SCOPES = os.environ.get(
'SPOTIFY_SCOPES',
'user-read-email user-read-private',
)