"""Spotify API Wrapper module.""" import base64 import datetime import gzip from io import StringIO import json import requests from requests import Request from requests import Session class SpotifyAPI(object): """Wrapper class to abstract loading data from Spotify API.""" _version = 'v1' _token = None _oauth_url = 'https://accounts.spotify.com/api/token' _resource_url = ( 'https://provider-api.spotify.com/{version}/analytics/' '{licensor_name}/enhanced/{resource_name}/{year}/{month}/{day}' '?version=4') _resource_url_by_country = ( 'https://provider-api.spotify.com/{version}/analytics/' '{licensor_name}/enhanced/{resource_name}' '/{year}/{month}/{day}/{country}?version=4') def __init__(self, client_id, client_secret, licensor_name, version=None): """Class constructor. Args: client_id (str): The account name. client_secret (str): The secret password. licensor_name (str): The licensor name, for example 'sony'. version (str): The api version. """ self.client_id = client_id self.client_secret = client_secret self.licensor_name = licensor_name self.version = version or self._version @property def token(self): """Spotify API oauth token. Requests oauth token from Spotify and caches it for the future uses. Returns: str: oauth token. """ if self._token is None: self._token = self._request_token() return self._token def _request_token(self): """Request oauth token from Spotify. Returns: str: oauth token. """ post_fields = { 'grant_type': 'client_credentials'} res = requests.post( self._oauth_url, data=post_fields, headers=self._get_auth_header_for_token()) if res.status_code != 200: raise requests.RequestException(response=res) return res.json()['access_token'] def _get_auth_header_for_token(self): """Return oauth header for requesting token. Returns: dict: The dict {'Authorization': 'Basic {}'}. """ credentials = '{}:{}'.format(self.client_id, self.client_secret) base64_output = base64.b64encode(credentials.encode()).decode() return {'Authorization': 'Basic {}'.format(base64_output)} def _get_auth_header_for_request(self): """Return oauth header for requesting data. Returns: dict: The dict {'Authorization': 'Bearer {token}'}. """ return {'Authorization': 'Bearer {}'.format(self.token)} def _request_to_file(self, fd, resource_name, date, version, country=None): """Request data from Spotify API and write raw gzipped response into fd. Args: fd (file): File object or writable buffer. resource_name (str): Name of resource to request. date (date or str): date instance or `YYYY-MM-DD` formatted string. version (str): API version. country (str): Optional country code argument. """ req = self._prepare_request( resource_name, date, version, country) res = Session().send(req, stream=True) res.raise_for_status() for chunk in res.iter_content(chunk_size=1024*1024): fd.write(chunk) fd.flush() def _request_to_json(self, resource_name, date, version, country=None): """Request data from Spotify API and return json object. Note that this method is extremely memory inefficient and `_request_to_file` should be preferred if possible. Args: resource_name (str): Name of resource to request. date (date or str): date instance or `YYYY-MM-DD` formatted string. version (str): API version. country (str): Optional country code argument. Returns: list: List of JSON objects from Spotify response. """ req = self._prepare_request( resource_name, date, version, country) res = Session().send(req) res.raise_for_status() res = StringIO(gzip.decompress(res.content).decode('utf-8')) return [json.loads(line) for line in res] def _prepare_request( self, resource_name, date, version, country=None): """Prepare Spotify API request. Args: resource_name (str): Name of resource to request. date (date or str): date instance or `YYYY-MM-DD` formatted string. version (str): API version. country (str): Optional country code argument. Returns: PreparedRequest: Prepared Request. """ year, month, day = self._parse_date(date) if country is not None: req = Request('GET', self._resource_url_by_country.format( licensor_name=self.licensor_name, version=version, resource_name=resource_name, year=year, month=month, day=day, country=country, token=self.token), headers=self._get_auth_header_for_request()) else: req = Request('GET', self._resource_url.format( licensor_name=self.licensor_name, version=version, resource_name=resource_name, year=year, month=month, day=day, token=self.token), headers=self._get_auth_header_for_request()) return req.prepare() def _parse_date(self, date): """Parse date into year, month, day tuple. Args: date (date or str): date instance or `YYYY-MM-DD` formatted string. Returns: tuple: year, month, day strings. """ if isinstance(date, datetime.date): return ( str(date.year), '{0:02d}'.format(date.month), '{0:02d}'.format(date.day)) elif isinstance(date, str): date = datetime.datetime.strptime(date, '%Y-%m-%d') return ( str(date.year), '{0:02d}'.format(date.month), '{0:02d}'.format(date.day)) else: raise ValueError( 'datetime.date or `YYYY-MM-DD` string expected. ' 'Got: {}'.format(date)) def get_tracks_to_file(self, fd, date): """Request Tracks data and write into file. Args: fd (file): File object or writable buffer. date (date or str): date instance or `YYYY-MM-DD` formatted string. """ self._request_to_file(fd, 'tracks', date, self.version) def get_tracks_to_json(self, date): """Request Tracks data and return json object. Args: date (date or str): date instance or `YYYY-MM-DD` formatted string. Returns: list: List of JSON objects from Spotify response. """ return self._request_to_json('tracks', date, self.version) def get_users_to_file(self, fd, date): """Request Users data and write into file. Args: fd (file): File object or writable buffer. date (date or str): date instance or `YYYY-MM-DD` formatted string. """ self._request_to_file(fd, 'users', date, self.version) def get_users_to_json(self, date): """Request Users data and return json object. Args: date (date or str): date instance or `YYYY-MM-DD` formatted string. Returns: list: List of JSON objects from Spotify response. """ return self._request_to_json('users', date, self.version) def get_streams_to_file(self, fd, date, country): """Request Streams data and write into file. Args: fd (file): File object or writable buffer. date (date or str): date instance or `YYYY-MM-DD` formatted string. country (str): Country code. """ self._request_to_file(fd, 'streams', date, self.version, country) def get_streams_to_json(self, date, country): """Request Streams data and return json object. Args: date (date or str): date instance or `YYYY-MM-DD` formatted string. country (str): Country code. Returns: list: List of JSON objects from Spotify response. """ return self._request_to_json( 'streams', date, self.version, country) def get_aggregated_streams_to_file(self, fd, date): """Request Aggregated Streams data and write into file. Args: fd (file): File object or writable buffer. date (date or str): date instance or `YYYY-MM-DD` formatted string. """ self._request_to_file(fd, 'aggregatedstreams', date, self.version) def get_aggregated_streams_to_json(self, date): """Request Aggregated Streams data and return json object. Args: date (date or str): date instance or `YYYY-MM-DD` formatted string. Returns: list: List of JSON objects from Spotify response. """ return self._request_to_json( 'aggregatedstreams', date, self.version) def get_sub_30_sec_streams_to_file(self, fd, date, country): """Request sub_30_second_streams data and write into file. Args: fd (file): File object or writable buffer. date (date or str): date instance or `YYYY-MM-DD` formatted string. country (str): Country code. """ self._request_to_file( fd, 'sub_30_sec_streams', date, self.version, country) def get_sub_30_sec_streams_to_json(self, date, country): """Request sub_30_second_streams data and return json object. Args: date (date or str): date instance or `YYYY-MM-DD` formatted string. country (str): Country code. Returns: list: List of JSON objects from Spotify response. """ return self._request_to_json( 'sub_30_sec_streams', date, self.version, country)