From 41aa93c3c7a37eeaaf79b4d35b3cc4c64b23cc57 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 10 Aug 2026 13:41:43 +0200 Subject: [PATCH] feat: expose recommendation streaming api --- backend/app/api/recommendations.py | 178 +++++++++++++++++++++++ backend/app/api/routes.py | 35 ++++- backend/app/main.py | 84 ++++++++++- backend/app/observability/logging.py | 30 ++++ backend/app/observability/timing.py | 3 + backend/tests/test_auth_routes.py | 29 +++- backend/tests/test_recommendation_api.py | 126 ++++++++++++++++ 7 files changed, 473 insertions(+), 12 deletions(-) create mode 100644 backend/app/api/recommendations.py create mode 100644 backend/app/observability/logging.py create mode 100644 backend/tests/test_recommendation_api.py diff --git a/backend/app/api/recommendations.py b/backend/app/api/recommendations.py new file mode 100644 index 0000000..4b36afb --- /dev/null +++ b/backend/app/api/recommendations.py @@ -0,0 +1,178 @@ +"""Authenticated recommendation streaming and playlist creation routes.""" + +import asyncio +from collections.abc import AsyncIterator, Callable +from typing import cast + +import structlog +from fastapi import APIRouter, HTTPException, Request +from fastapi.responses import StreamingResponse + +from app.adapters.spotify.errors import SpotifyError +from app.adapters.spotify.session import SpotifySession +from app.api.routes import resolve_session +from app.api.schemas import ( + NDJSON_CONTENT_TYPE, + DoneEvent, + ErrorEvent, + MetadataEvent, + PlaylistCreateRequest, + PlaylistCreateResponse, + RecommendationRequest, + StreamEvent, + TrackCard, + TrackEvent, + WarningEvent, +) +from app.config import Settings +from app.domain.models import ConversationTurn, PreviousRecommendation +from app.pipeline.event import ( + PipelineDoneEvent, + PipelineErrorEvent, + PipelineEvent, + PipelineMetadataEvent, + PipelineTrackEvent, + PipelineWarningEvent, +) +from app.pipeline.orchestrator import RecommendationPipeline +from app.ports.protocols import MusicCatalog, PlaylistWriter + +SpotifyClientFactory = Callable[[SpotifySession], MusicCatalog | PlaylistWriter] + +router = APIRouter() + + +@router.post("/api/recommendations") +async def recommendations( + request: Request, + payload: RecommendationRequest, +) -> StreamingResponse: + """Stream one authenticated discovery response as NDJSON.""" + resolved = resolve_session(request) + if resolved is None: + raise HTTPException(status_code=401, detail="Not authenticated") + + pipeline = cast(RecommendationPipeline, request.app.state.recommendation_pipeline) + factory = cast(SpotifyClientFactory, request.app.state.spotify_client_factory) + catalog = cast(MusicCatalog, factory(resolved.session)) + request_id = cast(str, request.state.request_id) + history = tuple(ConversationTurn(turn.role, turn.content) for turn in payload.history) + previous = tuple( + PreviousRecommendation( + item.rank, + item.track_id, + item.title, + tuple(item.artists), + ) + for item in payload.prior_recommendations + ) + + return StreamingResponse( + _stream_lines( + request, + pipeline, + resolved.session_id, + request_id, + catalog, + payload.query, + history, + previous, + ), + media_type=NDJSON_CONTENT_TYPE, + ) + + +@router.post("/api/playlists", response_model=PlaylistCreateResponse) +async def create_playlist( + request: Request, + payload: PlaylistCreateRequest, +) -> PlaylistCreateResponse: + """Create and fill one authenticated Spotify playlist.""" + resolved = resolve_session(request) + if resolved is None: + raise HTTPException(status_code=401, detail="Not authenticated") + settings = cast(Settings, request.app.state.settings) + factory = cast(SpotifyClientFactory, request.app.state.spotify_client_factory) + writer = cast(PlaylistWriter, factory(resolved.session)) + playlist_name = f"{settings.playlist_name_prefix} {payload.name}" + try: + playlist = await writer.create_playlist( + playlist_name, + "Music discovery selected by the listener.", + ) + await writer.add_tracks_to_playlist(playlist.id, payload.track_uris) + except (SpotifyError, ValueError) as error: + structlog.get_logger().warning("playlist_write_failed", error_type=type(error).__name__) + raise HTTPException(status_code=502, detail="Spotify playlist creation failed") from error + return PlaylistCreateResponse(url=playlist.url) + + +async def _stream_lines( + request: Request, + pipeline: RecommendationPipeline, + session_id: str, + request_id: str, + catalog: MusicCatalog, + query: str, + history: tuple[ConversationTurn, ...], + previous: tuple[PreviousRecommendation, ...], +) -> AsyncIterator[str]: + event_stream = pipeline.stream( + session_id, + request_id, + catalog, + query, + history, + previous, + ) + try: + async for event in event_stream: + if await request.is_disconnected(): + return + yield f"{_to_wire_event(event).model_dump_json()}\n" + except asyncio.CancelledError: + raise + except Exception as error: + structlog.get_logger().exception( + "recommendation_stream_failed", + error_type=type(error).__name__, + ) + if not await request.is_disconnected(): + failure = ErrorEvent( + code="recommendation_failed", + message="Recommendation could not be completed.", + ) + yield f"{failure.model_dump_json()}\n" + finally: + await event_stream.aclose() + + +def _to_wire_event(event: PipelineEvent) -> StreamEvent: + if isinstance(event, PipelineMetadataEvent): + return MetadataEvent( + request_id=event.request_id, + intent_summary=event.intent_summary, + candidate_count=event.candidate_count, + ) + if isinstance(event, PipelineTrackEvent): + track = event.track + return TrackEvent( + rank=event.rank, + track=TrackCard( + id=track.id, + uri=track.uri, + title=track.title, + artists=list(track.artists), + album_name=track.album_name, + album_art_url=track.album_art_url, + external_url=track.external_url, + ), + justification=event.justification, + ) + if isinstance(event, PipelineWarningEvent): + return WarningEvent(code=event.code, message=event.message) + if isinstance(event, PipelineErrorEvent): + return ErrorEvent(code=event.code, message=event.message) + if isinstance(event, PipelineDoneEvent): + return DoneEvent(track_count=event.track_count, total_ms=event.total_ms) + raise AssertionError("Unhandled pipeline event") diff --git a/backend/app/api/routes.py b/backend/app/api/routes.py index 7b72413..aab0682 100644 --- a/backend/app/api/routes.py +++ b/backend/app/api/routes.py @@ -1,5 +1,6 @@ """HTTP routes for Spotify login and session management.""" +from dataclasses import dataclass from typing import cast import httpx2 @@ -7,7 +8,7 @@ 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.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession from app.config import Settings SESSION_COOKIE_NAME = "discovery_session" @@ -15,6 +16,14 @@ 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.""" @@ -65,12 +74,10 @@ async def callback( @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: + resolved = resolve_session(request) + if resolved is None: return JSONResponse({"detail": "Not authenticated"}, status_code=401) - return JSONResponse({"display_name": session.display_name}) + return JSONResponse({"display_name": resolved.session.display_name}) @router.post("/api/auth/logout", status_code=204) @@ -95,3 +102,19 @@ def logout(request: Request) -> 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) diff --git a/backend/app/main.py b/backend/app/main.py index ceea6ad..ee1ea42 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,16 +1,24 @@ -"""Application factory and wiring. No logic lives here.""" +"""Application factory and dependency wiring.""" from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path import httpx2 +from anthropic import AsyncAnthropic from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from app.adapters.spotify.session import PendingLogins, SessionStore +from app.adapters.anthropic.llm import AnthropicRecommender +from app.adapters.spotify.auth import TokenSet, refresh_access_token +from app.adapters.spotify.client import SpotifyClient +from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession +from app.api.recommendations import router as recommendations_router from app.api.routes import router from app.config import AppMode, Settings, settings +from app.observability.logging import configure_logging +from app.observability.timing import RequestTimingMiddleware +from app.pipeline.orchestrator import RecommendationPipeline FRONTEND_DIST = Path(__file__).parent / "static" @@ -21,11 +29,11 @@ def create_app( ) -> FastAPI: """Build the FastAPI app: API routes plus the built SPA on one port.""" active_settings = application_settings or settings + configure_logging(active_settings.app_mode) @asynccontextmanager async def lifespan(application: FastAPI) -> AsyncIterator[None]: - if active_settings.app_mode is AppMode.LIVE and not active_settings.spotify_client_id: - raise RuntimeError("SPOTIFY_CLIENT_ID is required in live mode") + _validate_live_settings(active_settings) async with httpx2.AsyncClient( timeout=active_settings.spotify_timeout_seconds, transport=http_transport, @@ -34,7 +42,33 @@ def create_app( application.state.session_store = SessionStore() application.state.pending_logins = PendingLogins() application.state.settings = active_settings - yield + application.state.seed_session_id = None + anthropic_client = AsyncAnthropic( + api_key=active_settings.anthropic_api_key or "unused-demo-key" + ) + application.state.anthropic = anthropic_client + application.state.recommendation_pipeline = RecommendationPipeline( + AnthropicRecommender(anthropic_client, active_settings), + active_settings, + ) + + def spotify_client_factory(session: SpotifySession) -> SpotifyClient: + return SpotifyClient(http, session, active_settings) + + application.state.spotify_client_factory = spotify_client_factory + if ( + active_settings.app_mode is AppMode.LIVE + and active_settings.spotify_seed_refresh_token + ): + application.state.seed_session_id = await _install_seed_session( + http, + application.state.session_store, + active_settings, + ) + try: + yield + finally: + await anthropic_client.close() app = FastAPI( title="discovery-by-llm", @@ -42,12 +76,14 @@ def create_app( redoc_url=None, lifespan=lifespan, ) + app.add_middleware(RequestTimingMiddleware) @app.get("/api/health") def health() -> dict[str, str]: return {"status": "ok", "mode": active_settings.app_mode} app.include_router(router) + app.include_router(recommendations_router) if FRONTEND_DIST.is_dir(): app.mount("/", StaticFiles(directory=FRONTEND_DIST, html=True), name="spa") @@ -55,3 +91,41 @@ def create_app( app = create_app() + + +def _validate_live_settings(application_settings: Settings) -> None: + if application_settings.app_mode is not AppMode.LIVE: + return + if not application_settings.spotify_client_id: + raise RuntimeError("SPOTIFY_CLIENT_ID is required in live mode") + if not application_settings.anthropic_api_key: + raise RuntimeError("ANTHROPIC_API_KEY is required in live mode") + + +async def _install_seed_session( + http: httpx2.AsyncClient, + session_store: SessionStore, + application_settings: Settings, +) -> str: + tokens = await refresh_access_token( + http, + client_id=application_settings.spotify_client_id, + tokens=TokenSet( + access_token="seed-bootstrap", + refresh_token=application_settings.spotify_seed_refresh_token, + expires_at=0.0, + ), + ) + bootstrap_session = SpotifySession(tokens=tokens, account_id="", display_name="") + current_user = await SpotifyClient( + http, + bootstrap_session, + application_settings, + ).fetch_current_user() + return session_store.create( + SpotifySession( + tokens=bootstrap_session.tokens, + account_id=current_user.account_id, + display_name=current_user.display_name, + ) + ) diff --git a/backend/app/observability/logging.py b/backend/app/observability/logging.py new file mode 100644 index 0000000..1dba958 --- /dev/null +++ b/backend/app/observability/logging.py @@ -0,0 +1,30 @@ +"""Configure structlog for machine-readable live logs and readable development logs.""" + +import logging + +import structlog + +from app.config import AppMode + + +def configure_logging(app_mode: AppMode) -> None: + """Install the process logging pipeline for the selected runtime mode.""" + logging.basicConfig(level=logging.INFO, format="%(message)s") + renderer: structlog.types.Processor + if app_mode is AppMode.LIVE: + renderer = structlog.processors.JSONRenderer() + else: + renderer = structlog.dev.ConsoleRenderer(colors=False) + structlog.configure( + processors=[ + structlog.contextvars.merge_contextvars, + structlog.processors.add_log_level, + structlog.processors.TimeStamper(fmt="iso", utc=True), + structlog.processors.StackInfoRenderer(), + structlog.processors.format_exc_info, + renderer, + ], + wrapper_class=structlog.make_filtering_bound_logger(logging.INFO), + logger_factory=structlog.PrintLoggerFactory(), + cache_logger_on_first_use=True, + ) diff --git a/backend/app/observability/timing.py b/backend/app/observability/timing.py index 315761c..cd1843b 100644 --- a/backend/app/observability/timing.py +++ b/backend/app/observability/timing.py @@ -8,6 +8,7 @@ from typing import cast import structlog from starlette.types import ASGIApp, Receive, Scope, Send +from structlog.contextvars import bind_contextvars, reset_contextvars @dataclass @@ -71,10 +72,12 @@ class RequestTimingMiddleware: counters = RequestCounters() counter_token = _COUNTERS.set(counters) request_token = _REQUEST_ID.set(request_id) + logging_tokens = bind_contextvars(request_id=request_id) try: await self.app(scope, receive, send) finally: self._log_completion(scope, request_id, started_at, counters) + reset_contextvars(**logging_tokens) _COUNTERS.reset(counter_token) _REQUEST_ID.reset(request_token) diff --git a/backend/tests/test_auth_routes.py b/backend/tests/test_auth_routes.py index 523654b..c1c067f 100644 --- a/backend/tests/test_auth_routes.py +++ b/backend/tests/test_auth_routes.py @@ -90,5 +90,32 @@ def test_spotify_failure_during_callback_redirects_to_login_error() -> None: assert response.headers["location"] == "/?login=error" +def test_seed_session_authenticates_requests_without_a_cookie() -> None: + async def spotify_handler(request: httpx2.Request) -> httpx2.Response: + if request.url.host == "accounts.spotify.com": + return httpx2.Response(200, json={"access_token": "seed-access", "expires_in": 3600}) + assert request.url.path == "/v1/me" + return httpx2.Response(200, json={"id": "seed-account", "display_name": "Seed Listener"}) + + app = create_app( + application_settings=Settings( + app_mode=AppMode.LIVE, + spotify_client_id="client-id", + anthropic_api_key="test-key", + spotify_seed_refresh_token="seed-refresh", + ), + http_transport=httpx2.MockTransport(spotify_handler), + ) + with TestClient(app) as client: + response = client.get("/api/auth/me") + + assert response.status_code == 200 + assert response.json() == {"display_name": "Seed Listener"} + + def _live_settings() -> Settings: - return Settings(app_mode=AppMode.LIVE, spotify_client_id="client-id") + return Settings( + app_mode=AppMode.LIVE, + spotify_client_id="client-id", + anthropic_api_key="test-key", + ) diff --git a/backend/tests/test_recommendation_api.py b/backend/tests/test_recommendation_api.py new file mode 100644 index 0000000..f5e83ce --- /dev/null +++ b/backend/tests/test_recommendation_api.py @@ -0,0 +1,126 @@ +"""HTTP contract tests for recommendation and playlist routes.""" + +import time +from collections.abc import AsyncGenerator + +from fastapi.testclient import TestClient +from pydantic import TypeAdapter + +from app.adapters.spotify.auth import TokenSet +from app.adapters.spotify.session import SessionStore, SpotifySession +from app.api.routes import SESSION_COOKIE_NAME +from app.api.schemas import StreamEvent +from app.domain.models import ConversationTurn, CreatedPlaylist, PreviousRecommendation, Track +from app.main import create_app +from app.pipeline.event import PipelineDoneEvent, PipelineMetadataEvent, PipelineTrackEvent +from app.ports.protocols import MusicCatalog + + +class FakePipeline: + """Emit one complete deterministic stream.""" + + async def stream( + self, + session_id: str, + request_id: str, + catalog: MusicCatalog, + query: str, + history: tuple[ConversationTurn, ...], + previous_recommendations: tuple[PreviousRecommendation, ...], + ) -> AsyncGenerator[PipelineMetadataEvent | PipelineTrackEvent | PipelineDoneEvent]: + """Yield metadata, one track, and completion.""" + yield PipelineMetadataEvent(request_id, "A focused test request.", 35) + yield PipelineTrackEvent(1, _track(), "It fits the requested focus.") + yield PipelineDoneEvent(1, 4) + + +class FakePlaylistWriter: + """Capture playlist writes without external calls.""" + + def __init__(self) -> None: + """Create an empty write trace.""" + self.name: str | None = None + self.track_uris: list[str] = [] + + async def create_playlist(self, name: str, description: str) -> CreatedPlaylist: + """Record the prefixed name and return a stable playlist.""" + self.name = name + return CreatedPlaylist("playlist", "https://open.spotify.com/playlist/playlist") + + async def add_tracks_to_playlist(self, playlist_id: str, track_uris: list[str]) -> None: + """Record the ordered track URIs.""" + self.track_uris = track_uris + + +def test_recommendations_stream_lines_validate_against_frozen_schemas() -> None: + app = create_app() + with TestClient(app) as client: + _authenticate(client, session_store=app.state.session_store) + app.state.recommendation_pipeline = FakePipeline() + + response = client.post( + "/api/recommendations", + json={"schema_version": 1, "query": "focused electronic music"}, + ) + + adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent) + events = [adapter.validate_json(line) for line in response.text.splitlines()] + assert response.status_code == 200 + assert response.headers["content-type"] == "application/x-ndjson" + assert [event.type for event in events] == ["metadata", "track", "done"] + + +def test_recommendations_require_a_valid_session_without_seed() -> None: + with TestClient(create_app()) as client: + response = client.post( + "/api/recommendations", + json={"schema_version": 1, "query": "focused electronic music"}, + ) + + assert response.status_code == 401 + assert response.json() == {"detail": "Not authenticated"} + + +def test_playlist_endpoint_prefixes_name_and_adds_tracks() -> None: + app = create_app() + writer = FakePlaylistWriter() + with TestClient(app) as client: + _authenticate(client, session_store=app.state.session_store) + app.state.spotify_client_factory = lambda session: writer + + response = client.post( + "/api/playlists", + json={ + "schema_version": 1, + "name": "Night drive", + "track_uris": ["spotify:track:track"], + }, + ) + + assert response.status_code == 200 + assert response.json() == {"url": "https://open.spotify.com/playlist/playlist"} + assert writer.name == "discovery-by-llm Night drive" + assert writer.track_uris == ["spotify:track:track"] + + +def _authenticate(client: TestClient, session_store: SessionStore) -> None: + session_id = session_store.create( + SpotifySession( + tokens=TokenSet("access", "refresh", time.monotonic() + 3600), + account_id="account", + display_name="Listener", + ) + ) + client.cookies.set(SESSION_COOKIE_NAME, session_id) + + +def _track() -> Track: + return Track( + id="track", + uri="spotify:track:track", + title="Test Track", + artists=("Test Artist",), + album_name="Test Album", + album_art_url=None, + external_url="https://open.spotify.com/track/track", + )