"""Spotify api client.""" import re import backoff import spotipy from requests.exceptions import ConnectionError as RequestsConnectionError, HTTPError # 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 SpotifyClient(spotipy.Spotify): """Extended Spotify client.""" @backoff.on_exception( backoff.expo, _BACKOFF_EXCEPTIONS, max_tries=3, giveup=_giveup ) def artist(self, artist_id): """Get Spotify catalog information for a single artist. Args: artist_id (str): The Spotify ID for the artist. Returns: dict: Artist object. """ return super().artist(artist_id) @backoff.on_exception( backoff.expo, _BACKOFF_EXCEPTIONS, max_tries=3, giveup=_giveup ) def search( self, q, limit=10, offset=0, search_type='track', market=None, locale=None ): """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. """ return self._get( 'search', q=q, limit=limit, offset=offset, type=search_type, market=market, locale=locale, ) @backoff.on_exception( backoff.expo, _BACKOFF_EXCEPTIONS, max_tries=3, giveup=_giveup ) def get_artists_by_ids(self, *, artist_ids: list[str]) -> dict[str, dict | None]: """Return artist objects keyed by id, None if not found. Args: artist_ids (list[str]): List of Spotify artist ids. Returns: dict[str, dict | 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}