import asyncio import base64 import time from typing import List, Union import aiohttp import config from server.core.cache import ListResult, SingleResult, cache_data, cache_request from server.core.constants import CollectionName from server.core.utils import ListResult as ChunkListResult from server.core.utils import get_id, make_request, request_in_chunks, retry from server.core.utils import spotify_pagination as pagination from server.spotify.exceptions import SpotifyError from server.spotify.utils import (compare_tracks_ids_data, get_playlist_image_url, get_track_image_url, replace_with_linked_data) class SpotifyClientCredentials: """Credentials for Spotify API""" OAUTH_URL = config.SPOTIFY_CLIENT_TOKEN_URL def __init__(self): """Credentials client for Spotify.""" self.client_id = config.SPOTIFY_CLIENT_ID self.client_secret = config.SPOTIFY_CLIENT_SECRET self.token_info = None async def get_access_token(self, session: aiohttp.ClientSession) -> str: """If a valid access token is in memory return it else fetch a new token.""" if self.token_info and not self._is_token_expired(): return self.token_info["access_token"] token_info = await self._get_access_token(session) token_info["expires_at"] = int(time.time()) + token_info["expires_in"] self.token_info = token_info return self.token_info["access_token"] async def _get_access_token(self, session: aiohttp.ClientSession) -> dict: """Gets client credentials access token.""" payload = {"grant_type": "client_credentials"} auth_header = base64.b64encode(f"{self.client_id}:{self.client_secret}".encode()) headers = {"Authorization": "Basic %s" % auth_header.decode()} return await make_request( session, self.OAUTH_URL, method="POST", data=payload, headers=headers, error_cls=SpotifyError, error_message="Credentials are not correct", ) def _is_token_expired(self) -> bool: if not self.token_info: return True now = int(time.time()) return self.token_info["expires_at"] - now < 60 class Spotify: """Client for Spotify API""" def __init__(self, session: aiohttp.ClientSession, credentials: SpotifyClientCredentials): """Create a Spotify API instance.""" self.prefix = config.SPOTIFY_CLIENT_BASE_URL self.credentials = credentials self.session = session async def _auth_headers(self) -> dict: token = await self.credentials.get_access_token(self.session) return {"Authorization": f"Bearer {token}"} async def _make_request(self, url: str, **kwargs) -> dict: url = f"{self.prefix}/{url}" headers = await self._auth_headers() headers["Content-Type"] = "application/json" return await make_request(self.session, url, params=kwargs, headers=headers, error_cls=SpotifyError) @retry() async def _get(self, url: str, **kwargs) -> dict: return await self._make_request(url, **kwargs) @cache_request(collection_name=CollectionName.SPOTIFY_CACHE) async def cached_get(self, url: str, **kwargs) -> dict: return await self._get(url, **kwargs) @cache_data(result_handler=SingleResult(CollectionName.SPOTIFY_TRACK)) async def track(self, track_id: str, market: str = None) -> dict: """Returns a single track given the track's ID, URI or URL. Args: track_id: Spotify URI, URL or ID. market: ISO 3166-1 alpha-2 country code. Returns: Spotify track metadata. """ params = {"market": market} if market else {} track_data = await self._get(f"v1/tracks/{track_id}", **params) if track_data["id"] != track_id: track_data = replace_with_linked_data(track_data) return track_data @cache_data(result_handler=ListResult(CollectionName.SPOTIFY_TRACK)) @request_in_chunks(50, result_type=ChunkListResult) async def _tracks(self, tracks: List[str], market: str = None) -> List[dict]: """Returns a list of tracks given a list of track IDs.""" params = {"market": market} if market else {} result = await self._get("v1/tracks/?ids=" + ",".join(tracks), **params) tracks_list = [track for track in result.get("tracks", []) if track] if tracks_list: return compare_tracks_ids_data(tracks, tracks_list) return tracks_list @cache_data( result_handler=ListResult( key_options=( (CollectionName.SPOTIFY_TRACK_ISRC, lambda x: x["external_ids"]["isrc"]), (CollectionName.SPOTIFY_TRACK, get_id), ) ) ) @request_in_chunks(1, result_type=ChunkListResult) async def _tracks_by_isrc(self, isrc: List[str], market: str = None) -> dict: """Returns a list of tracks given a list of track IDs.""" data = await self._search(q=f"isrc:{isrc[0]}", market=market, limit=1) # remove pagination fields return data["tracks"].get("items", []) async def tracks(self, track_ids: List[str] = None, market: str = None, by_isrc: bool = False) -> dict: """Returns a list of tracks given a list of track IDs, URIs, or URLs Args: track_ids: A list of spotify URIs, URLs or IDs or ISRC. market: ISO 3166-1 alpha-2 country code. by_isrc: Get tracks by track IDs/URIs or by ISRC. Returns: Dict with list of tracks data in JSON format. """ if by_isrc: tracks_list = await self._tracks_by_isrc(track_ids, market=market) else: tracks_list = await self._tracks(track_ids, market=market) return {"tracks": tracks_list} async def tracks_images(self, tracks: List[str], image_size: int, market: str = None) -> List[dict]: """Returns tracks images urls given a list of track IDs, URIs, or URLs Args: tracks: A list of spotify URIs, URLs or IDs. image_size: Required image size in pixels. market: ISO 3166-1 alpha-2 country code. Returns: A list of track image url in JSON format. """ tracks_list = await self._tracks(tracks, market=market) return [{"id": get_id(track), "image_url": get_track_image_url(track, image_size)} for track in tracks_list] @pagination(100) @cache_request(collection_name=CollectionName.SPOTIFY_PLAYLIST_TRACKS) async def playlist_tracks( self, playlist_id: str, market: str = None, fields: List[str] or None = None, additional_types: List[str] or None = None, offset: int = 0, limit: int = 100, ) -> dict: """Method returns playlist`s list of tracks. Args: playlist_id: Playlist ID. market: an ISO 3166-1 alpha-2 country code. fields: Fields filter. additional_types: Track and / or episode. offset: Page offset. limit: Page limit (100 max value). Returns: Playlist tracklist. """ params = {"offset": offset, "limit": limit} if market: params["market"] = market if fields: params["fields"] = fields if additional_types: params["additional_types"] = ",".join(additional_types) return await self._get(f"v1/playlists/{playlist_id}/tracks", **params) @cache_data(result_handler=SingleResult(CollectionName.SPOTIFY_PLAYLIST)) async def playlist( self, playlist_id: str, market: str = None, fields: List[str] or None = None, additional_types: List[str] or None = None, ) -> dict: """Returns playlist data by ID. Args: playlist_id: Playlist ID. market: ISO 3166-1 alpha-2 country code. fields: A list of additional fields. additional_types: Track and / or episode. Returns: Playlist data in JSON format. """ params = {"market": market} if market else {} if fields: params["fields"] = fields if additional_types: params["additional_types"] = ",".join(additional_types) return await self._get(f"v1/playlists/{playlist_id}", **params) @cache_data(result_handler=ListResult(CollectionName.SPOTIFY_PLAYLIST)) @request_in_chunks(result_type=ChunkListResult) async def playlists( self, playlists: List[str], market: str = None, fields: List[str] or None = None, additional_types: List[str] or None = None, ) -> List[dict]: """Returns playlist data by ID. Args: playlists: A list of playlist IDs. market: ISO 3166-1 alpha-2 country code. fields: A list of additional fields. additional_types: Track and / or episode. Returns: List of playlist data in JSON format. """ result = await asyncio.gather( *[self.playlist(playlist_id, market, fields, additional_types) for playlist_id in playlists] ) return [item for item in result if item is not None] async def playlists_images(self, playlists: List[str], market: str = None): """Get catalog playlist images urls associated with the identifiers provided. Args: playlists: A list of playlist IDs. market: ISO 3166-1 alpha-2 country code. Returns: A list of playlist image urls in JSON format. """ playlists_data = await self.playlists(playlists, market) return [{"id": playlist["id"], "image_url": get_playlist_image_url(playlist)} for playlist in playlists_data] async def _search( self, q: str, market: str, item_type: Union[List[str], str] = "track", include_external: str or None = None, offset: int = 0, limit: int = 10, ) -> dict: """Search Spotify metadata.""" params = { "q": q, "type": ",".join(item_type) if isinstance(item_type, list) else item_type, "offset": offset, "limit": limit, } if market: params["market"] = market if include_external: params["include_external"] = include_external return await self._get("v1/search", **params) @cache_request(collection_name=CollectionName.SPOTIFY_SEARCH) async def search( self, q: str, market: str, item_type: List[str], include_external: str or None = None, offset: int = 0, limit: int = 10, ) -> dict: """Search Spotify metadata. Args: q: Search query. market: ISO 3166-1 alpha-2 country code or the string from_token. item_type: Item types to return. Could be 'artist', 'album', 'track' or 'playlist'. include_external: If include_external=audio is specified the response will include any relevant audio content that is hosted externally. offset: Index of the first item to return. limit: Number of items to return. Returns: Spotify metadata. """ return await self._search(q, market, item_type, include_external, offset, limit) @cache_data(result_handler=SingleResult(CollectionName.SPOTIFY_ALBUM)) async def album(self, album_id: str, market: str = None) -> dict: """Method returns a single album given the album's ID :param album_id: Album ID. :param market: an ISO 3166-1 alpha-2 country code. :return: Album data in JSON format """ params = {"market": market} if market else {} return await self._get("v1/albums/" + album_id, **params) @cache_request(collection_name=CollectionName.SPOTIFY_ALBUM_TRACKS) @pagination(50) async def album_tracks(self, album_id: str, market: str = None, offset: int = 0, limit: int = 50) -> dict: """Method returns a single album given the album's ID :param album_id: Album ID. :param market: an ISO 3166-1 alpha-2 country code. :param offset: Page offset. :param limit: Page limit (50 max value). :return: Album data in JSON format """ params = {"offset": offset, "limit": limit} if market: params["market"] = market return await self._get(f"v1/albums/{album_id}/tracks", **params) @cache_data(result_handler=ListResult(CollectionName.SPOTIFY_ALBUM, result_items_key="albums")) @request_in_chunks(20) async def albums(self, albums: List[str], market: str = None) -> dict: """Returns a list of tracks given a list of track IDs, URIs, or URLs :param albums: A list of Spotify album IDs. :param market: an ISO 3166-1 alpha-2 country code. :return: List of albums data. """ params = {"market": market} if market else {} return await self._get("v1/albums/?ids=" + ",".join(albums), **params) @cache_data(result_handler=SingleResult(CollectionName.SPOTIFY_ARTIST)) async def artist(self, artist_id: str) -> dict: """Method returns a single artist given the artist's ID :param artist_id: Artist ID. :return: Artist data in JSON format """ return await self._get("v1/artists/" + artist_id) @cache_data(result_handler=ListResult(CollectionName.SPOTIFY_ARTIST, result_items_key="artists")) @request_in_chunks(50) async def artists(self, artists: List[str]) -> dict: """Returns a list of tracks given a list of track IDs, URIs, or URLs :param artists: A list of Spotify artist IDs. :return: List of artists data. """ return await self._get("v1/artists/?ids=" + ",".join(artists)) @cache_data(result_handler=SingleResult(CollectionName.SPOTIFY_USER)) async def user(self, user_id: str) -> dict: """Method returns a single user given the user's ID :param user_id: User ID. :return: User data in JSON format """ return await self._get("v1/users/" + user_id) @cache_request(collection_name=CollectionName.SPOTIFY_MARKETS) async def markets(self) -> list: """Return list of supported markets.""" result = (await self._get("v1/markets")).get("markets", []) return [m.lower() for m in result] + [""] async def check_health(self): response = await self._make_request("v1/users/spotify") return bool(response)