diff --git a/app/spotify.py b/app/spotify.py index a484646..2021498 100644 --- a/app/spotify.py +++ b/app/spotify.py @@ -1,31 +1,69 @@ +"""Handle connections with the Spotify API.""" import base64 import requests +import urllib +from datetime import datetime, timedelta +from typing import List + + +AUTH_URL = "https://accounts.spotify.com/api/token" +API_URL = "https://api.spotify.com/v1" + + +def requires_token(method: callable, retries: int = 2): + """Decorator that handles refreshing the token. + + Args: + method: function to decorate + retries: how many times to retry creating the token + """ + + def inner(ref, *args, **kwargs): + for _ in range(retries): + if ref._access_token and datetime.now() < ref._expiration_time: + break + + ref._refresh_token() + + method(ref, *args, **kwargs) + + return inner class CredentialError(Exception): + """Thrown when invalid credentials are passed to Spotify API.""" + pass class Spotify: - def __init__(self, client_id, client_secret): - self.client_id = client_id - self.client_secret = client_secret - self.access_token = "" - self.token_type = "" - self.expiration_time = 0 - self.scope = "" + """Represents a connection object with the Spotify API.""" - def request_token(self): + def __init__(self, client_id: str, client_secret: str): + """Initialize a new Spotify object. + + Args: + client_id: client id for the API + client_secret: client secret for the API + """ + self._client_id = client_id + self._client_secret = client_secret + + self._access_token = "" + self._token_type = "" + self._expiration_time = 0 + self._scope = "" + + def _refresh_token(self): + """Refresh the current token, or create a new one.""" b64_encoded = base64.b64encode( - "{}:{}".format(self.client_id, self.client_secret).encode() + f"{self.client_id}:{self.client_secret}".encode() ) - headers = {"Authorization": "Basic {}".format(b64_encoded.decode("ascii"))} + headers = {"Authorization": f"Basic {b64_encoded.decode('ascii')}"} data = {"grant_type": "client_credentials"} - post = requests.post( - "https://accounts.spotify.com/api/token", headers=headers, data=data - ) + post = requests.post(AUTH_URL, headers=headers, data=data) status_code = post.status_code if status_code == 401: @@ -34,22 +72,27 @@ class Spotify: if status_code == 200: self.access_token = post.json()["access_token"] self.token_type = post.json()["token_type"] - self.expiration_time = post.json()["expires_in"] + self.expiration_time = datetime.now() + timedelta(seconds=post.json()["expires_in"]) self.scope = post.json()["scope"] - def search(self, search_term, types): - if not self.access_token: - self.request_token() + # TODO raise errors on failure + @requires_token + def search(self, search_term: str, types: List[str]): + """Search the API for entries. + + Args: + search_term: the searm term to use + types: which object types to search for + """ encoded_search_term = urllib.parse.quote(search_term) - full_url = "https://api.spotify.com/v1/search?q={}&type={}".format( - encoded_search_term, ",".join(types) + types_str = ",".join(types) + + r = requests.get( + f"{API_URL}/search?q={encoded_search_term}&type={types_str}", + headers={"Authorization": "Bearer " + self.access_token}, ) - get = requests.get( - full_url, headers={"Authorization": "Bearer " + self.access_token} - ) - - results = get.json() + results = r.json() return results diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a8f43fe --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,2 @@ +[tool.black] +line-length = 79