import re import backoff from pydantic import BaseModel from requests.exceptions import ConnectionError as RequestsConnectionError from requests.exceptions import HTTPError from spotipy import Spotify as SpotipyClient # Artist ids are 22 characters, assuming a theoretical query string max of 2048 # characters, this is conservative to a bit less than half that limit. CHUNK_SIZE = 40 # Spotify ID regex. The API will throw errors if the ID is not valid, so we # pre-validate them. SPOTIFY_ID_REGEX = re.compile(r"^[0-9A-Za-z]{22}$") _BACKOFF_EXCEPTIONS = ( RequestsConnectionError, # handles connection resets (errno 104) HTTPError, # handles 429 and 5xx ) def _giveup(e): """Only retry on connection errors and 429/5xx HTTP errors.""" if isinstance(e, RequestsConnectionError): return False return e.response.status_code != 429 and ( e.response.status_code < 500 or e.response.status_code >= 600 ) def chunked(strings: list[str], size: int = CHUNK_SIZE): """Yield successive n-sized chunks from a list of strings.""" for i in range(0, len(strings), size): yield strings[i : i + size] def valid_spotify_id(string: str) -> bool: """Check if a string is a valid Spotify ID (base 62 string).""" return bool(SPOTIFY_ID_REGEX.fullmatch(string)) class SpotifyArtist(BaseModel): identifier: str name: str followers: int | None genres: list[str] | None url: str image: str | None class SpotifyClient(SpotipyClient): """Extended Spotify client.""" @backoff.on_exception( backoff.expo, _BACKOFF_EXCEPTIONS, max_tries=3, giveup=_giveup ) def artist(self, artist_id) -> SpotifyArtist | None: """Get Spotify catalog information for a single artist. Args: artist_id (str): The Spotify ID for the artist. Returns: SpotifyArtist | None: Artist object. """ result = super().artist(artist_id) return SpotifyArtist( identifier=result["id"], name=result["name"], followers=result.get("followers", {}).get("total", None), genres=result.get("genres", None), url=result["external_urls"]["spotify"], image=result["images"][0]["url"] if result["images"] else None, ) @backoff.on_exception( backoff.expo, _BACKOFF_EXCEPTIONS, max_tries=3, giveup=_giveup ) def search( self, q, limit=10, offset=0, type="track", market=None, locale=None ) -> list[SpotifyArtist]: """Search for an item. Args: q (str): The search query. limit (int): The number of items to return. offset (int): The index of the first item to return. search_type (str): The type of item to return. One of 'artist', 'album', 'track' or 'playlist' market (str): An ISO 3166-1 alpha-2 country code or the string from_token. locale (str): The language of the response. Undocumented feature. """ response = self._get( "search", q=q, limit=limit, offset=offset, type=type, market=market, locale=locale, ) return [ SpotifyArtist( identifier=artist["id"], name=artist["name"], followers=artist["followers"]["total"], genres=artist["genres"], url=artist["external_urls"]["spotify"], image=artist["images"][0]["url"] if artist["images"] else None, ) for artist in response["artists"]["items"] ] @backoff.on_exception( backoff.expo, _BACKOFF_EXCEPTIONS, max_tries=3, giveup=_giveup ) def get_artists_by_ids( self, *, artist_ids: list[str] ) -> dict[str, SpotifyArtist | None]: """Return artist objects keyed by id, None if not found. Args: artist_ids (list[str]): List of Spotify artist ids. Returns: dict[str, SpotifyArtist | None]: Mapping of artist id to artist object or None. """ valid_artist_ids = [ artist_id for artist_id in artist_ids if valid_spotify_id(artist_id) ] artists = [] for chunk in chunked(valid_artist_ids): results = self.artists(chunk) assert results artists.extend(results["artists"]) artist_map = { artist["id"]: artist for artist in artists if artist and "id" in artist } return {artist_id: artist_map.get(artist_id, None) for artist_id in artist_ids}