Refactor images

Use weblink parsing and factor out group_by_type helper.
mopidy · kingosticks · Apr 16, 2024 · Mar 15, 2024 · Mar 26, 2024 · Apr 16, 2024
commit 393b46f1c6997fe64a0eccd00ba6769919c66733
diff --git a/src/mopidy_spotify/images.py b/src/mopidy_spotify/images.py
@@ -1,12 +1,17 @@
-import itertools
 import logging
-import operator
-import urllib.parse
 
 from mopidy_spotify.browse import BROWSE_DIR_URIS
 from mopidy_spotify.translator import web_to_image
+from mopidy_spotify.utils import group_by_type
+from mopidy_spotify.web import LinkType, WebLink
 
 _API_MAX_IDS_PER_REQUEST = 50
+SUPPORTED_TYPES = (
+    LinkType.TRACK,
+    LinkType.ALBUM,
+    LinkType.ARTIST,
+    LinkType.PLAYLIST,
+)
 
 _cache = {}  # (type, id) -> [Image(), ...]
 
@@ -15,113 +20,92 @@
 
 def get_images(web_client, uris):
     result = {}
-    uri_type_getter = operator.itemgetter("type")
-    uris = (_parse_uri(u) for u in uris)
-    uris = sorted((u for u in uris if u), key=uri_type_getter)
-    for uri_type, group in itertools.groupby(uris, uri_type_getter):
+    links = (_parse_uri(u) for u in uris)
+    for link_type, link_group in group_by_type(links):
         batch = []
-        for uri in group:
-            if uri["key"] in _cache:
-                result[uri["uri"]] = _cache[uri["key"]]
-            elif uri_type == "playlist":
-                result.update(_process_uri(web_client, uri))
+        for link in link_group:
+            key = _make_cache_key(link)
+            if key in _cache:
+                result[link.uri] = _cache[key]
+            elif link_type == LinkType.PLAYLIST:
+                result.update(_process_one(web_client, link))
             else:
-                batch.append(uri)
+                batch.append(link)
                 if len(batch) >= _API_MAX_IDS_PER_REQUEST:
-                    result.update(_process_uris(web_client, uri_type, batch))
+                    result.update(_process_many(web_client, link_type, batch))
                     batch = []
-        result.update(_process_uris(web_client, uri_type, batch))
+        result.update(_process_many(web_client, link_type, batch))
     return result
 
 
+def _make_cache_key(link):
+    return (link.type, link.id)
+
+
 def _parse_uri(uri):
     if uri in BROWSE_DIR_URIS:
         return None  # These are internal to the extension.
     try:
-        parsed_uri = urllib.parse.urlparse(uri)
-        uri_type, uri_id = None, None
-
-        match parsed_uri.scheme:
-            case "spotify":
-                match parsed_uri.path.split(":"):
-                    case uri_type, uri_id, *_:
-                        pass
-                    case _:
-                        raise ValueError("Too few arguments")  # noqa: TRY301
-            case "http" | "https":
-                if parsed_uri.netloc in ("open.spotify.com", "play.spotify.com"):
-                    uri_type, uri_id = parsed_uri.path.split("/")[1:3]
-
-        supported_types = ("track", "album", "artist", "playlist")
-        if uri_type:
-            if uri_type not in supported_types:
-                logger.warning(f"Unsupported image type '{uri_type}' in {uri!r}")
-                return None
-            if uri_id:
-                return {
-                    "uri": uri,
-                    "type": uri_type,
-                    "id": uri_id,
-                    "key": (uri_type, uri_id),
-                }
-        raise ValueError("Unknown error")  # noqa: TRY301
+        link = WebLink.from_uri(uri)
+        if link.type not in SUPPORTED_TYPES:
+            raise ValueError(f"Unsupported image type '{link.type}' in {uri!r}")  # noqa: TRY301
+        if not link.id:
+            raise ValueError("ID missing")  # noqa: TRY301
     except Exception as e:
         logger.exception(f"Could not parse {uri!r} as a Spotify URI ({e!s})")  # noqa: TRY401
+        return None
+
+    return link
 
 
-def _process_uri(web_client, uri):
-    data = web_client.get(f"{uri['type']}s/{uri['id']}")
-    _cache[uri["key"]] = tuple(web_to_image(i) for i in data.get("images") or [])
-    return {uri["uri"]: _cache[uri["key"]]}
+def _process_one(web_client, link):
+    data = web_client.get(f"{link.type}s/{link.id}")
+    key = _make_cache_key(link)
+    _cache[key] = tuple(web_to_image(i) for i in data.get("images") or [])
+    return {link.uri: _cache[key]}
 
 
-def _process_uris(  # noqa: C901
+def _process_many(  # noqa: C901
     web_client,
-    uri_type,
-    uris,
+    link_type,
+    links,
 ):
     result = {}
-    ids = [u["id"] for u in uris]
-    ids_to_uris = {u["id"]: u for u in uris}
-
-    if not uris:
+    if not links:
         return result
 
-    data = web_client.get(uri_type + "s", params={"ids": ",".join(ids)})
-    for item in (
-        data.get(
-            uri_type + "s",
-        )
-        or []
-    ):
+    ids = [u.id for u in links]
+    ids_to_links = {u.id: u for u in links}
+
+    data = web_client.get(link_type + "s", params={"ids": ",".join(ids)})
+    for item in data.get(link_type + "s") or []:
         if not item:
             continue
 
         if "linked_from" in item:
             item_id = item["linked_from"].get("id")
         else:
             item_id = item.get("id")
-        uri = ids_to_uris.get(item_id)
-        if not uri:
+        link = ids_to_links.get(item_id)
+        if not link:
             continue
 
-        if uri["key"] not in _cache:
-            if uri_type == "track":
+        key = _make_cache_key(link)
+        if key not in _cache:
+            if link_type == LinkType.TRACK:
                 if "album" not in item:
                     continue
-                album = _parse_uri(item["album"].get("uri"))
-                if not album:
+                album_link = _parse_uri(item["album"].get("uri"))
+                if not album_link:
                     continue
-                album_key = album["key"]
+                album_key = _make_cache_key(album_link)
                 if album_key not in _cache:
                     _cache[album_key] = tuple(
                         web_to_image(i) for i in item["album"].get("images") or []
                     )
-                _cache[uri["key"]] = _cache[album_key]
+                _cache[key] = _cache[album_key]
             else:
-                _cache[uri["key"]] = tuple(
-                    web_to_image(i) for i in item.get("images") or []
-                )
-        result[uri["uri"]] = _cache[uri["key"]]
+                _cache[key] = tuple(web_to_image(i) for i in item.get("images") or [])
+        result[link.uri] = _cache[key]
 
     return result
diff --git a/src/mopidy_spotify/utils.py b/src/mopidy_spotify/utils.py
@@ -1,5 +1,7 @@
 import contextlib
+import itertools
 import logging
+import operator
 import time
 
 import requests
@@ -33,3 +35,9 @@ def time_logger(name, level=TRACE):
 
 def flatten(list_of_lists):
     return [item for sublist in list_of_lists for item in sublist]
+
+
+def group_by_type(links):
+    link_type_getter = operator.attrgetter("type")
+    links = sorted((u for u in links if u), key=link_type_getter)
+    return itertools.groupby(links, link_type_getter)
diff --git a/src/mopidy_spotify/web.py b/src/mopidy_spotify/web.py
@@ -8,7 +8,7 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from email.utils import parsedate_to_datetime
-from enum import Enum, unique
+from enum import StrEnum, auto, unique
 from http import HTTPStatus
 
 import requests
@@ -271,7 +271,7 @@ def _parse_retry_after(self, response):
 
 
 @unique
-class ExpiryStrategy(Enum):
+class ExpiryStrategy(StrEnum):
     FORCE_FRESH = "force-fresh"
     FORCE_EXPIRED = "force-expired"
 
@@ -554,12 +554,12 @@ def get_track(self, web_link):
 
 
 @unique
-class LinkType(Enum):
-    TRACK = "track"
-    ALBUM = "album"
-    ARTIST = "artist"
-    PLAYLIST = "playlist"
-    YOUR = "your"
+class LinkType(StrEnum):
+    TRACK = auto()
+    ALBUM = auto()
+    ARTIST = auto()
+    PLAYLIST = auto()
+    YOUR = auto()
 
 
 @dataclass

diff --git a/tests/test_images.py b/tests/test_images.py
@@ -245,11 +245,9 @@ def test_max_50_ids_per_request(web_client_mock, img_provider):
     "uri",
     [
         "foo:bar",
-        "spotify:baz",
-        "spotify:artist",
-        "spotify:album",
-        "spotify:user",
+        "spotify:user:bob:starred",
         "spotify:playlist",
+        "spotify:your:fish",
     ],
 )
 def test_invalid_uri(img_provider, caplog, uri):
@@ -259,14 +257,11 @@ def test_invalid_uri(img_provider, caplog, uri):
     assert f"Could not parse '{uri}' as a Spotify URI" in caplog.text
 
 
-@pytest.mark.parametrize(
-    "uri", ["spotify:dog:cat", "spotify:your:fish", "spotify:top:hat"]
-)
-def test_unsupported_image_type(img_provider, caplog, uri):
+def test_unsupported_image_type(img_provider, caplog):
     with caplog.at_level(5):
-        result = img_provider.get_images([uri])
+        result = img_provider.get_images(["spotify:your:fish"])
     assert result == {}
-    assert f"Unsupported image type '{uri.split(':')[1]}'" in caplog.text
+    assert "Unsupported image type 'your'" in caplog.text
 
 
 @pytest.mark.parametrize(

diff --git a/tests/test_web.py b/tests/test_web.py
@@ -1444,10 +1444,16 @@ def test_weblink_from_uri_playlist(uri, id_, owner):
         ("spotify:user:alice:track:foo"),
         ("local:user:alice:playlist:foo"),
         ("spotify:track:foo:bar"),
-        ("spotify:album:"),
         ("https://yahoo.com/playlist/foo"),
         ("https://play.spotify.com/foo"),
         ("total/junk"),
+        ("foo:bar"),
+        ("spotify:baz"),
+        ("spotify:artist"),
+        ("spotify:album"),
+        ("spotify:user"),
+        ("spotify:playlist"),
+        ("spotify:playlist:"),
     ],
 )
 def test_weblink_from_uri_raises(uri):