Added auto-refreshing Spotify token

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):
+    """Represents a connection object with the Spotify API."""
-        self.client_id = client_id
-        self.client_secret = client_secret
-        self.access_token = ""
-        self.token_type = ""
-        self.expiration_time = 0
-        self.scope = ""
 
-    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(
+        post = requests.post(AUTH_URL, headers=headers, data=data)
-            "https://accounts.spotify.com/api/token", 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):
+        # TODO raise errors on failure
-        if not self.access_token:
-            self.request_token()
 
+    @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(
+        types_str = ",".join(types)
-            encoded_search_term, ",".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(
+        results = r.json()
-            full_url, headers={"Authorization": "Bearer " + self.access_token}
-        )
-
-        results = get.json()
 
         return results

pyproject.toml

@@ -0,0 +1,2 @@
+[tool.black]
+line-length = 79