97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""HTTP routes for Spotify login and session management."""
|
|
|
|
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
|
|
from app.config import Settings
|
|
|
|
SESSION_COOKIE_NAME = "discovery_session"
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@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."""
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
session = session_store.get(session_id) if session_id is not None else None
|
|
if session is None:
|
|
return JSONResponse({"detail": "Not authenticated"}, status_code=401)
|
|
return JSONResponse({"display_name": 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)
|