Fall back to using Spotipy

Would rather have async I/O, but there are too many issues with the data model of other APIs.

spotify_actions/__init__.py

@@ -0,0 +1,3 @@
+"""
+"Actions" library for automating Spotify workflows
+"""

spotify_actions/echo.py

@@ -2,12 +2,12 @@
 Methods for printing results to console; primarily useful when developing/debugging pipelines to
 check results before committing.
 """
-from typing import AsyncIterable
+from typing import Iterable
 
-from spotify import Album
+from spotify_model.album import SearchAlbum
 
 
-async def echo_album(albums: AsyncIterable[Album]) -> None:
+def echo_album(albums: Iterable[SearchAlbum]) -> None:
     "Print album metadata"
-    async for album in albums:
+    for album in albums:
         print(album.name)

spotify_actions/search.py

@@ -1,13 +1,22 @@
 """
 Utility methods for the Spotify query API
 """
-from typing import AsyncIterable
+from typing import Any, Dict, Iterable, cast
 
-from spotify import Album, Client
+from spotify_model.album import SearchAlbum
+from spotipy import Spotify
+
+from .util import exhaust
 
 
-async def search_album(client: Client, search_str: str) -> AsyncIterable[Album]:
-    "Search for a specific album by name"
-    results = await client.search(search_str, types=["album"])
-    for album in results.albums:
-        yield album
+def search_album(client: Spotify, search_str: str) -> Iterable[SearchAlbum]:
+    "Display albums from a search string"
+
+    def _search(limit: int, offset: int) -> Dict[str, Any]:
+        return cast(
+            Dict[str, Any],
+            client.search(search_str, limit=limit, offset=offset, type="album")["albums"],
+        )
+
+    for item in exhaust(_search):
+        yield SearchAlbum(**item)

spotify_actions/util.py

@@ -1,80 +1,42 @@
 """
 Utility methods for working with the Spotify API
 """
-from asyncio import AbstractEventLoop
-from logging import getLogger
+from math import ceil
 from pathlib import Path
-from typing import Any, Awaitable, Dict, Optional
+from typing import Any, Dict, Iterable, Protocol
 
 import yaml
-from aiohttp import (ClientSession, TraceConfig, TraceRequestEndParams,
-                     TraceResponseChunkReceivedParams)
-from spotify import Client, HTTPClient
+from spotipy import Spotify, SpotifyClientCredentials
 
-LOGGER = getLogger(__name__)
+DEFAULT_LIMIT = 50
 
 
-class UnescapeHTTPClient(HTTPClient):
-    "Disable URL-escaping queries"
-
-    def search(
-        self,
-        q: str,
-        query_type: str = "track,playlist,artist,album",
-        market: str = "US",
-        limit: int = 20,
-        offset: int = 0,
-        include_external: Optional[str] = None,
-    ) -> Awaitable:
-        route = self.route("GET", "/search")
-        payload: Dict[str, Any] = {
-            # The important bit is not quoting here
-            "q": q,
-            "type": query_type,
-            "limit": limit,
-            "offset": offset,
-        }
-
-        if market:
-            payload["market"] = market
-
-        if include_external is not None:
-            payload["include_external"] = include_external
-
-        return self.request(route, params=payload)
-
-
-class UnescapeClient(Client):
-
-    _default_http_client = UnescapeHTTPClient
-
-
-def read_credentials(
-    path: Path, loop: AbstractEventLoop, trace_config: Optional[TraceConfig] = None
-) -> Client:
+def read_credentials(path: Path) -> Spotify:
     "Read credentials from a YAML file and construct a Spotify client"
 
-    if trace_config:
-
-        class TracingHTTPClient(UnescapeHTTPClient):
-            def __init__(self, *args, **kwargs):
-                super().__init__(*args, **kwargs)
-
-                prev_session = self._session
-                self._session = ClientSession(loop=loop, trace_configs=[trace_config])
-                loop.run_until_complete(prev_session.close())
-
-        class TracingClient(Client):
-
-            _default_http_client = TracingHTTPClient
-
-        client_cls = TracingClient
-
-    else:
-        client_cls = Client
-
     with open(path, "r") as credentials_file:
         credentials = yaml.safe_load(credentials_file)
-        return client_cls(
-            credentials["client_id"], credentials["client_secret"], loop=loop
-        )
+        credentials_manager = SpotifyClientCredentials(credentials["client_id"], credentials["client_secret"])
+        return Spotify(client_credentials_manager=credentials_manager)
+
+
+class Paginated(Protocol):
+    "Protocol definition for functions that will be provided to the `exhaust` handler"
+
+    def __call__(self, limit: int, offset: int) -> Dict[str, Any]:
+        ...
+
+
+def exhaust(function: Paginated, limit: int = DEFAULT_LIMIT) -> Iterable[Dict[str, Any]]:
+    "Exhaust a function that returns a pagination object"
+    response = function(limit=limit, offset=0)
+    total = response["total"]
+    limit = response["limit"]
+
+    for item in response["items"]:
+        yield item
+
+    for i in range(1, ceil(total / limit)):
+        response = function(limit=limit, offset=limit * i)
+        for item in response["items"]:
+            yield item