discovery-by-llm/backend/app/adapters/spotify/login.py
2026-08-10 11:31:06 +02:00

66 lines
2 KiB
Python

"""Spotify login workflow independent of HTTP routing."""
import secrets
import httpx2
from app.adapters.spotify.auth import (
build_authorize_url,
derive_code_challenge,
exchange_authorization_code,
generate_code_verifier,
)
from app.adapters.spotify.client import SpotifyClient
from app.adapters.spotify.errors import SpotifyError
from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession
from app.config import Settings
def begin_login(settings: Settings, pending_logins: PendingLogins) -> str:
"""Store a pending PKCE login and return its Spotify authorization URL."""
state = secrets.token_urlsafe(32)
code_verifier = generate_code_verifier()
pending_logins.add(state, code_verifier)
return build_authorize_url(
settings.spotify_client_id,
settings.spotify_redirect_uri,
state,
derive_code_challenge(code_verifier),
)
async def complete_login(
http: httpx2.AsyncClient,
settings: Settings,
pending_logins: PendingLogins,
session_store: SessionStore,
code: str,
state: str,
) -> str | None:
"""Complete a pending Spotify login and return its application session ID."""
code_verifier = pending_logins.pop(state)
if code_verifier is None:
return None
try:
tokens = await exchange_authorization_code(
http,
client_id=settings.spotify_client_id,
redirect_uri=settings.spotify_redirect_uri,
code=code,
code_verifier=code_verifier,
)
bootstrap_session = SpotifySession(tokens=tokens, account_id="", display_name="")
current_user = await SpotifyClient(
http,
bootstrap_session,
settings,
).fetch_current_user()
session = SpotifySession(
tokens=bootstrap_session.tokens,
account_id=current_user.account_id,
display_name=current_user.display_name,
)
return session_store.create(session)
except (SpotifyError, ValueError):
return None