Added auto-refreshing Spotify token

.woodpecker.yml

@@ -2,7 +2,7 @@ pipeline:
     # =====TESTING=====
     test-backend:
         # Alpine version doesn't have make
-        image: python:3.8
+        image: python:3.9
         pull: true
         group: test
         commands:
@@ -23,7 +23,7 @@ pipeline:
 
     # =====LINTING=====
     lint-backend:
-        image: python:3.8
+        image: python:3.9
         group: lint
         commands:
             - make lint

Makefile

@@ -3,7 +3,7 @@ PYTHON  := python3
 # This can't contain spaces (I think)
 VENV    := .venv
 # Minimum % coverage for tests to succeed
-MIN_COV := 0
+MIN_COV := 50
 # Directory name for the frontend
 WEB_DIR := web
 
@@ -55,7 +55,7 @@ test: venv
 ## Starting the server
 ### Run the Quart server
 run: venv
-	@ '$(VENV)'/bin/python app
+	@ QUART_ENV=development '$(VENV)'/bin/python app
 .PHONY: run
 
 

app/__main__.py

@@ -1,7 +1,9 @@
-"""Main entrypoint for the program."""
+"""Entrypoint for the program."""
 from quart import Quart
+from app.api import api_bp
 
 app = Quart("jos", static_folder="web/dist", static_url_path="/")
+app.register_blueprint(api_bp, url_prefix="/api")
 
 
 @app.route("/", methods=["GET"], defaults={"path": ""})
@@ -11,4 +13,5 @@ async def frontend(path):
     return await app.send_static_file("index.html")
 
 
+if __name__ == "__main__":
     app.run(host="0.0.0.0")

app/api/__init__.py

@@ -0,0 +1,8 @@
+"""Module containing all Flask-related code."""
+from quart import Blueprint
+from .search import search_bp
+
+
+# Main blueprint exposing entire API
+api_bp = Blueprint("api", __name__)
+api_bp.register_blueprint(search_bp, url_prefix="/search")

app/api/search.py

@@ -0,0 +1,11 @@
+"""Handles the search endpoint."""
+from quart import Blueprint
+
+
+search_bp = Blueprint("search", __name__)
+
+
+@search_bp.route("/", methods=["GET"])
+async def search_spotify():
+    """Search the Spotify API."""
+    return "yeet"

app/spotify.py

@@ -0,0 +1,98 @@
+"""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:
+    """Represents a connection object with the Spotify API."""
+
+    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(
+            f"{self.client_id}:{self.client_secret}".encode()
+        )
+
+        headers = {"Authorization": f"Basic {b64_encoded.decode('ascii')}"}
+        data = {"grant_type": "client_credentials"}
+
+        post = requests.post(AUTH_URL, headers=headers, data=data)
+        status_code = post.status_code
+
+        if status_code == 401:
+            raise CredentialError("Provided credentials are not correct.")
+
+        if status_code == 200:
+            self.access_token = post.json()["access_token"]
+            self.token_type = post.json()["token_type"]
+            self.expiration_time = datetime.now() + timedelta(seconds=post.json()["expires_in"])
+            self.scope = post.json()["scope"]
+
+        # 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)
+        types_str = ",".join(types)
+
+        r = requests.get(
+            f"{API_URL}/search?q={encoded_search_term}&type={types_str}",
+            headers={"Authorization": "Bearer " + self.access_token},
+        )
+
+        results = r.json()
+
+        return results

tests/test_succeed.py

@@ -1,3 +0,0 @@
-def test_succeed():
-    """Placeholder test to make CI succeed."""
-    pass