120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""HTTP routes for Spotify login and session management."""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import cast
|
|
|
|
import httpx2
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import JSONResponse, RedirectResponse, Response
|
|
|
|
from app.adapters.spotify.login import begin_login, complete_login
|
|
from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession
|
|
from app.config import Settings
|
|
|
|
SESSION_COOKIE_NAME = "discovery_session"
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolvedSession:
|
|
"""A session and stable cache key selected for one request."""
|
|
|
|
session_id: str
|
|
session: SpotifySession
|
|
|
|
|
|
@router.get("/api/auth/login")
|
|
def login(request: Request) -> RedirectResponse:
|
|
"""Start Spotify Authorization Code with PKCE login."""
|
|
application_settings = cast(Settings, request.app.state.settings)
|
|
pending_logins = cast(PendingLogins, request.app.state.pending_logins)
|
|
authorize_url = begin_login(application_settings, pending_logins)
|
|
return RedirectResponse(authorize_url, status_code=307)
|
|
|
|
|
|
@router.get("/callback")
|
|
async def callback(
|
|
request: Request,
|
|
code: str | None = None,
|
|
state: str | None = None,
|
|
error: str | None = None,
|
|
) -> RedirectResponse:
|
|
"""Complete Spotify login and establish an opaque cookie session."""
|
|
if error is not None or code is None or state is None:
|
|
return _login_error_redirect()
|
|
|
|
application_settings = cast(Settings, request.app.state.settings)
|
|
http = cast(httpx2.AsyncClient, request.app.state.http)
|
|
pending_logins = cast(PendingLogins, request.app.state.pending_logins)
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
session_id = await complete_login(
|
|
http,
|
|
application_settings,
|
|
pending_logins,
|
|
session_store,
|
|
code,
|
|
state,
|
|
)
|
|
if session_id is None:
|
|
return _login_error_redirect()
|
|
|
|
response = RedirectResponse("/", status_code=307)
|
|
response.set_cookie(
|
|
SESSION_COOKIE_NAME,
|
|
session_id,
|
|
httponly=True,
|
|
samesite="lax",
|
|
path="/",
|
|
secure=application_settings.session_cookie_secure,
|
|
)
|
|
return response
|
|
|
|
|
|
@router.get("/api/auth/me")
|
|
def current_session(request: Request) -> JSONResponse:
|
|
"""Return the display name for a valid application session."""
|
|
resolved = resolve_session(request)
|
|
if resolved is None:
|
|
return JSONResponse({"detail": "Not authenticated"}, status_code=401)
|
|
return JSONResponse({"display_name": resolved.session.display_name})
|
|
|
|
|
|
@router.post("/api/auth/logout", status_code=204)
|
|
def logout(request: Request) -> Response:
|
|
"""Remove the current application session and clear its cookie."""
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
if session_id is not None:
|
|
session_store.remove(session_id)
|
|
|
|
application_settings = cast(Settings, request.app.state.settings)
|
|
response = Response(status_code=204)
|
|
response.delete_cookie(
|
|
SESSION_COOKIE_NAME,
|
|
path="/",
|
|
secure=application_settings.session_cookie_secure,
|
|
httponly=True,
|
|
samesite="lax",
|
|
)
|
|
return response
|
|
|
|
|
|
def _login_error_redirect() -> RedirectResponse:
|
|
return RedirectResponse("/?login=error", status_code=307)
|
|
|
|
|
|
def resolve_session(request: Request) -> ResolvedSession | None:
|
|
"""Resolve the cookie session or the installed live seed session."""
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
cookie_session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if cookie_session_id is not None:
|
|
cookie_session = session_store.get(cookie_session_id)
|
|
if cookie_session is not None:
|
|
return ResolvedSession(cookie_session_id, cookie_session)
|
|
|
|
seed_session_id = cast(str | None, getattr(request.app.state, "seed_session_id", None))
|
|
seed_session = session_store.get(seed_session_id) if seed_session_id is not None else None
|
|
if seed_session is None or seed_session_id is None:
|
|
return None
|
|
return ResolvedSession(seed_session_id, seed_session)
|