spotify_auth_search.py

import requests
import base64
import datetime
from urllib.parse import urlencode

client_id = '' # https://developer.spotify.com/dashboard/applications/
client_secret = ''


# auth
client_creds = f'{client_id}:{client_secret}'
client_creds_b64 = base64.b64encode(client_creds.encode()) # base64.b64decode(client_creds_b64)
token_url = 'https://accounts.spotify.com/api/token'
token_data = {
  'grant_type': 'client_credentials'
}
token_headers = {
  'Authorization': f'Basic {client_creds_b64.decode()}'
}
r = requests.post(token_url, data=token_data, headers=token_headers)
valid_request = r.status_code in range(200, 299)
if valid_request:
  token_response_data = r.json()
  access_token = token_response_data['access_token']
  expires_in = token_response_data['expires_in']
  now = datetime.datetime.now()
  expires = now + datetime.timedelta(seconds=expires_in)
  did_expire = expires < now
print(access_token)



class SpotifyAPI(object):
  access_token = None
  access_token_expires = datetime.datetime.now()
  access_token_did_expire = True
  client_id = None
  client_secret = None
  token_url = 'https://accounts.spotify.com/api/token'
  search_endpoint = 'https://api.spotify.com/'
  version = 'v1'

  def __init__(self, client_id, client_secret, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.client_id = client_id
    self.client_secret = client_secret

  def get_client_credentials(self):
    client_id = self.client_id
    client_secret = self.client_secret
    if client_id == None or client_secret == None:
      raise Exception('You must set client_id and client_secret')
    client_creds = f'{client_id}:{client_secret}'
    client_creds_b64 = base64.b64encode(client_creds.encode()) # base64.b64decode(client_creds_b64)
    return client_creds_b64.decode()

  def get_token_headers(self):
    client_creds_b64 = self.get_client_credentials()
    return {'Authorization': f'Basic {client_creds_b64}',}

  def get_token_data(self):
    return {'grant_type': 'client_credentials'}

  def perform_auth(self):
    token_url = self.token_url
    token_data = self.get_token_data()
    token_headers = self.get_token_headers()
    r = requests.post(token_url, data=token_data, headers=token_headers)
    if r.status_code not in range(200, 299):
      raise Exception('Could not authenticate client.')
    data = r.json()
    now = datetime.datetime.now()
    expires = now + datetime.timedelta(seconds=data['expires_in'])
    self.access_token = data['access_token']
    self.access_token_expires = expires
    self.access_token_did_expire = expires < now
    return True

  def get_access_token(self):
    token = self.access_token
    expires = self.access_token_expires
    now = datetime.datetime.now()
    if expires < now:
      self.perform_auth()
      return self.get_access_token()
    elif token == None:
      self.perform_auth()
      return self.get_access_token()
    return token

  def get_resource_header(self):
    access_token = self.get_access_token()
    headers = {
      'Authorization': f'Bearer {access_token}'
    }
    return headers

  def base_search(self, query_params):
    headers = self.get_resource_header()
    endpoint = f'{self.search_endpoint}{self.version}/search'
    lookup_url = f'{endpoint}?{query_params}'
    r = requests.get(lookup_url, headers=headers)
    if r.status_code not in range(200, 299):
      return {}
    return r.json()

  def search(self, query=None, operator=None, operator_query=None, search_type='artist'):
    if query == None:
      raise Exception('A query is required')
    if isinstance(query, dict):
      query = ' '.join([f'{k}:{v}' for k,v in query.items()])
    if operator != None and operator_query != None:
      if operator.lower() == "or" or operator.lower() == "not":
        operator = operator.upper()
        if isinstance(operator_query, str):
          query = f"{query} {operator} {operator_query}"
    query_params = urlencode({'q': query, 'type': search_type.lower()})
    return self.base_search(query_params)

  def get_resource(self, lookup_id, resource_type='albums'):
    endpoint = f'{self.search_endpoint}{self.version}/{resource_type}/{lookup_id}/'
    headers = self.get_resource_header()
    r = requests.get(endpoint, headers=headers)
    if r.status_code not in range(200, 299):
      return {}
    return r.json()

  def get_track(self, lookup_id):
    return self.get_resource(lookup_id, resource_type='tracks')

  def get_album(self, lookup_id):
    return self.get_resource(lookup_id, resource_type='albums')
    
  def get_artist(self, lookup_id):
    return self.get_resource(lookup_id, resource_type='artists')

  # def basic_search(self, query, search_type='artist'):
  #   headers = self.get_resource_header()
  #   endpoint = f'{self.search_endpoint}{self.version}/search'
  #   data = urlencode({'q': query, 'type': search_type.lower()})
  #   lookup_url = f'{endpoint}?{data}'
  #   r = requests.get(lookup_url, headers=headers)
  #   if r.status_code not in range(200, 299):
  #     return {}
  #   return r.json()

spotify = SpotifyAPI(client_id, client_secret)
spotify.perform_auth()
print(spotify.access_token)