Wrap Spotify API in helper class to handle token refreshes

Author: linuxdaemon

plugins/spotify.py

@@ -1,87 +1,105 @@
 import re
-from datetime import datetime
+from datetime import datetime, timedelta
+from threading import RLock
 
 import requests
 from requests import HTTPError
 from requests.auth import HTTPBasicAuth
+from yarl import URL
 
 from cloudbot import hook
 
-api_url = "https://api.spotify.com/v1/search?"
-token_url = "https://accounts.spotify.com/api/token"
-spuri = 'spotify:{}:{}'
-access_token = ""
-expires_at = datetime.min
-
-spotify_re = re.compile(r'(spotify:(track|album|artist|user):([a-zA-Z0-9]+))',
-                        re.I)
-http_re = re.compile(r'(open\.spotify\.com/(track|album|artist|user)/'
-                     '([a-zA-Z0-9]+))', re.I)
-
-
-def sprequest(bot, params, alturl=None):
-    global access_token, expires_at
-    if alturl is None:
-        alturl = api_url
-    if datetime.now() >= expires_at:
-        basic_auth = HTTPBasicAuth(
-            bot.config.get("api_keys", {}).get("spotify_client_id"),
-            bot.config.get("api_keys", {}).get("spotify_client_secret"))
-        gtcc = {"grant_type": "client_credentials"}
-        auth = requests.post(token_url, data=gtcc, auth=basic_auth).json()
-        if 'access_token' in auth.keys():
-            access_token = auth["access_token"]
-            expires_at = datetime.fromtimestamp(datetime.now().timestamp() +
-                                                auth["expires_in"])
-    headers = {'Authorization': 'Bearer ' + access_token}
-    return requests.get(alturl, params=params, headers=headers)
+api = None
 
+spotify_re = re.compile(
+    r'(spotify:(track|album|artist|user):([a-zA-Z0-9]+))', re.I
+)
+http_re = re.compile(
+    r'(open\.spotify\.com/(track|album|artist|user)/([a-zA-Z0-9]+))', re.I
+)
 
-@hook.command('spotify', 'sptrack')
-def spotify(bot, text, reply):
-    """<song> - Search Spotify for <song>"""
-    params = {"q": text.strip(), "offset": 0, "limit": 1, "type": "track"}
+TYPE_MAP = {
+    'artist': 'artists',
+    'album': 'albums',
+    'track': 'tracks',
+}
+
+
+class SpotifyAPI:
+    api_url = URL("https://api.spotify.com/v1")
+    token_refresh_url = URL("https://accounts.spotify.com/api/token")
+
+    def __init__(self, client_id=None, client_secret=None):
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._access_token = None
+        self._token_expires = datetime.min
+        self._lock = RLock()  # Make sure only one requests is parsed at a time
+
+    def request(self, endpoint, params=None):
+        with self._lock:
+            if datetime.now() >= self._token_expires:
+                self._refresh_token()
+
+            r = requests.get(
+                self.api_url / endpoint, params=params, headers={'Authorization': 'Bearer ' + self._access_token}
+            )
+            r.raise_for_status()
+
+            return r
+
+    def search(self, params):
+        return self.request('search', params)
 
-    request = sprequest(bot, params)
+    def _refresh_token(self):
+        with self._lock:
+            basic_auth = HTTPBasicAuth(self._client_id, self._client_secret)
+            gtcc = {"grant_type": "client_credentials"}
+            r = requests.post(self.token_refresh_url, data=gtcc, auth=basic_auth)
+            r.raise_for_status()
+            auth = r.json()
+            self._access_token = auth["access_token"]
+            self._token_expires = datetime.now() + timedelta(seconds=auth["expires_in"])
+
+
+def _search(text, _type, reply):
+    params = {"q": text.strip(), "offset": 0, "limit": 1, "type": _type}
 
     try:
-        request.raise_for_status()
+        request = api.search(params)
     except HTTPError as e:
         reply("Could not get track information: {}".format(e.request.status_code))
         raise
 
-    if request.status_code != requests.codes.ok:
-        return "Could not get track information: {}".format(
-            request.status_code)
+    return request.json()[TYPE_MAP[_type]]["items"][0]
+
+
+@hook.onload
+def create_api(bot):
+    keys = bot.config['api_keys']
+    client_id = keys['spotify_client_id']
+    client_secret = keys['spotify_client_secret']
+    global api
+    api = SpotifyAPI(client_id, client_secret)
 
-    data = request.json()["tracks"]["items"][0]
+
+@hook.command('spotify', 'sptrack')
+def spotify(text, reply):
+    """<song> - Search Spotify for <song>"""
+    data = _search(text, "track", reply)
 
     try:
         return "\x02{}\x02 by \x02{}\x02 - {} / {}".format(
-            data["artists"][0]["name"], data["external_urls"]["spotify"],
-            data["name"], data["uri"])
+            data["name"], data["artists"][0]["name"],
+            data["external_urls"]["spotify"], data["uri"])
     except IndexError:
         return "Unable to find any tracks!"
 
 
 @hook.command("spalbum")
-def spalbum(bot, text, reply):
+def spalbum(text, reply):
     """<album> - Search Spotify for <album>"""
-    params = {"q": text.strip(), "offset": 0, "limit": 1, "type": "album"}
-
-    request = sprequest(bot, params)
-
-    try:
-        request.raise_for_status()
-    except HTTPError as e:
-        reply("Could not get album information: {}".format(e.request.status_code))
-        raise
-
-    if request.status_code != requests.codes.ok:
-        return "Could not get album information: {}".format(
-            request.status_code)
-
-    data = request.json()["albums"]["items"][0]
+    data = _search(text, "album", reply)
 
     try:
         return "\x02{}\x02 by \x02{}\x02 - {} / {}".format(
@@ -92,22 +110,9 @@ def spalbum(bot, text, reply):
 
 
 @hook.command("spartist", "artist")
-def spartist(bot, text, reply):
+def spartist(text, reply):
     """<artist> - Search Spotify for <artist>"""
-    params = {"q": text.strip(), "offset": 0, "limit": 1, "type": "artist"}
-
-    request = sprequest(bot, params)
-    try:
-        request.raise_for_status()
-    except HTTPError as e:
-        reply("Could not get artist information: {}".format(e.request.status_code))
-        raise
-
-    if request.status_code != requests.codes.ok:
-        return "Could not get artist information: {}".format(
-            request.status_code)
-
-    data = request.json()["artists"]["items"][0]
+    data = _search(text, "artist", reply)
 
     try:
         return "\x02{}\x02 - {} / {}".format(
@@ -118,15 +123,14 @@ def spartist(bot, text, reply):
 
 @hook.regex(http_re)
 @hook.regex(spotify_re)
-def spotify_url(bot, match):
-    api_method = {'track': 'tracks', 'album': 'albums', 'artist': 'artists'}
+def spotify_url(match):
     _type = match.group(2)
     spotify_id = match.group(3)
-    # no error catching here, if the API is down fail silently
-    request = sprequest(bot, {}, 'http://api.spotify.com/v1/{}/{}'.format(
-        api_method[_type], spotify_id))
-    request.raise_for_status()
+
+    request = api.request("{}/{}".format(TYPE_MAP[_type], spotify_id))
+
     data = request.json()
+
     if _type == "track":
         name = data["name"]
         artist = data["artists"][0]["name"]