import abc from typing import Any, ClassVar from pydantic import SecretStr from resonance_engine.dsp.enums import DSPId, DSPResource from resonance_engine.dsp.types import TokenResult class DSPBackend(abc.ABC): dsp_id: DSPId required_scopes: ClassVar[dict[DSPResource, str]] = {} def close(self) -> None: # noqa: B027 """Release any resources held by the DSP backend.""" def __enter__(self) -> DSPBackend: return self def __exit__(self, *_: object) -> None: self.close() def required_scope(self, resource: DSPResource) -> str | None: """OAuth scope this collection resource's endpoint requires, if any.""" return self.required_scopes.get(resource) def has_required_scope(self, resource: DSPResource, scope: str | None) -> bool: """True unless `resource` needs a scope the granted `scope` lacks.""" if scope is None: return True needed = self.required_scope(resource) return needed is None or needed in scope.split() @abc.abstractmethod def refresh_token(self, refresh_token: SecretStr) -> TokenResult: """Refresh OAuth token.""" @abc.abstractmethod def get_profile(self, access_token: SecretStr) -> dict[str, Any]: """Fetch profile data for the authenticated user.""" @abc.abstractmethod def get_top_artists(self, access_token: SecretStr) -> list[dict[str, Any]]: """Fetch top artists for the authenticated user.""" @abc.abstractmethod def get_top_tracks(self, access_token: SecretStr) -> list[dict[str, Any]]: """Fetch top tracks for the authenticated user.""" @abc.abstractmethod def get_recently_played( self, access_token: SecretStr, *, after: int | None = None ) -> list[dict[str, Any]]: """Fetch recently played tracks for the authenticated user.""" @abc.abstractmethod def get_playlists(self, access_token: SecretStr) -> list[dict[str, Any]]: """Fetch playlists for the authenticated user.""" @abc.abstractmethod def get_saved_albums(self, access_token: SecretStr) -> list[dict[str, Any]]: """Fetch saved albums for the authenticated user.""" @abc.abstractmethod def get_saved_tracks(self, access_token: SecretStr) -> list[dict[str, Any]]: """Fetch saved tracks for the authenticated user.""" @abc.abstractmethod def get_followed_artists(self, access_token: SecretStr) -> list[dict[str, Any]]: """Fetch followed artists for the authenticated user."""