"""Apple Artist Model.""" import datetime import jwt import requests from artist.config import secrets_manager_client from artist.constants import apple, errors from artist.response import create_error_response from artist.response import Response MUSIC_KIT_SECRET = '\n'.join([ '-----BEGIN PRIVATE KEY-----', secrets_manager_client.get_cred('MUSIC_KIT_PRIVATE_KEY') or '', '-----END PRIVATE KEY-----']) MUSIC_KIT_KEY_ID = secrets_manager_client.get_cred('MUSIC_KIT_KEY_ID') APPLE_MUSIC_TEAM_ID = secrets_manager_client.get_cred('APPLE_MUSIC_TEAM_ID') def get_auth_token(): """Generate an encrypted token using JWT specification. Required for all Apple Music API calls. Returns: str: Apple Music developer token. """ alg = apple.DSA time_now = datetime.datetime.now() time_expired = datetime.datetime.now() + datetime.timedelta(seconds=60) headers = { 'alg': alg, 'kid': MUSIC_KIT_KEY_ID } payload = { 'iss': APPLE_MUSIC_TEAM_ID, 'exp': int(time_expired.strftime('%s')), 'iat': int(time_now.strftime('%s')) } token = jwt.encode(payload, MUSIC_KIT_SECRET, algorithm=alg, headers=headers) return token.decode() def search(query, storefront, limit=None, offset=None): """Fetch a list of artists from Apple Music.""" auth_token = get_auth_token() response = requests.get( 'https://api.music.apple.com/v1/catalog/{}/search'.format(storefront), headers={'Authorization': 'Bearer {}'.format(auth_token)}, params={ 'term': query, 'types': 'artists', 'limit': limit, 'offset': offset } ) if response.status_code == 401: return create_error_response( code=errors.VALIDATION_ERROR, status=401, message='Failed to authenticate with Apple Music.' ) # Rate limited. if response.status_code == 429: return response response_json = response.json() results = response_json['results'] or [] if results: results = results['artists']['data'] return Response(message={'items': results})