Support lookups for 'your' URIs

This adds the ability to utilise the "spotify:your:tracks" and
"spotify:your:albums" URIs through library lookups.

Author: djmattyg007

mopidy_spotify/browse.py

@@ -218,25 +218,31 @@ def _browse_toplist(config, session, variant, region):
         return []
 
 
-def _browse_your_music(web_client, variant):
-    if not web_client.logged_in:
-        return []
+def _load_your_music(web_client, variant):
+    if web_client is None or not web_client.logged_in:
+        return
 
-    if variant in ("tracks", "albums"):
-        items = flatten(
-            [
-                page.get("items", [])
-                for page in web_client.get_all(
-                    f"me/{variant}",
-                    params={"market": "from_token", "limit": 50},
-                )
-                if page
-            ]
-        )
-        if variant == "tracks":
-            return list(translator.web_to_track_refs(items))
-        else:
-            return list(translator.web_to_album_refs(items))
+    if variant not in ("tracks", "albums"):
+        return
+
+    results = web_client.get_all(
+        f"me/{variant}",
+        params={"market": "from_token", "limit": 50},
+    )
+    for page in results:
+        if not page:
+            continue
+        items = page.get("items", [])
+        for item in items:
+            yield item
+
+
+def _browse_your_music(web_client, variant):
+    items = _load_your_music(web_client, variant)
+    if variant == "tracks":
+        return list(translator.web_to_track_refs(items))
+    elif variant == "albums":
+        return list(translator.web_to_album_refs(items))
     else:
         return []
 

mopidy_spotify/lookup.py

@@ -1,7 +1,8 @@
 import logging
 
 import spotify
-from mopidy_spotify import playlists, translator, utils, web
+from mopidy_spotify import browse, playlists, translator, utils
+from mopidy_spotify.web import LinkType, WebLink
 
 logger = logging.getLogger(__name__)
 
@@ -12,16 +13,18 @@
 
 def lookup(config, session, web_client, uri):
     try:
-        web_link = web.WebLink.from_uri(uri)
-        if web_link.type != web.LinkType.PLAYLIST:
+        web_link = WebLink.from_uri(uri)
+        if web_link.type not in (LinkType.PLAYLIST, LinkType.YOUR):
             sp_link = session.get_link(uri)
     except ValueError as exc:
         logger.info(f"Failed to lookup {uri!r}: {exc}")
         return []
 
     try:
-        if web_link.type == web.LinkType.PLAYLIST:
+        if web_link.type == LinkType.PLAYLIST:
             return _lookup_playlist(config, session, web_client, uri)
+        elif web_link.type == LinkType.YOUR:
+            return list(_lookup_your(config, session, web_client, uri))
         elif sp_link.type is spotify.LinkType.TRACK:
             return list(_lookup_track(config, sp_link))
         elif sp_link.type is spotify.LinkType.ALBUM:
@@ -92,3 +95,31 @@ def _lookup_playlist(config, session, web_client, uri):
     if playlist is None:
         raise spotify.Error("Playlist Web API lookup failed")
     return playlist.tracks
+
+
+def _lookup_your(config, session, web_client, uri):
+    parts = uri.replace("spotify:your:", "").split(":")
+    if len(parts) != 1:
+        return
+    variant = parts[0]
+
+    items = browse._load_your_music(web_client, variant)
+    if variant == "tracks":
+        for item in items:
+            # The extra level here is to also support "saved track objects".
+            web_track = item.get("track", item)
+            track = translator.web_to_track(
+                web_track, bitrate=config["bitrate"]
+            )
+            if track is not None:
+                yield track
+    elif variant == "albums":
+        for item in items:
+            # The extra level here is to also support "saved album objects".
+            web_album = item.get("album", item)
+            album_ref = translator.web_to_album_ref(web_album)
+            if album_ref is None:
+                continue
+            sp_link = session.get_link(album_ref.uri)
+            if sp_link.type is spotify.LinkType.ALBUM:
+                yield from _lookup_album(config, sp_link)

mopidy_spotify/web.py

@@ -486,6 +486,7 @@ class LinkType(Enum):
     ALBUM = "album"
     ARTIST = "artist"
     PLAYLIST = "playlist"
+    YOUR = "your"
 
 
 @dataclass
@@ -519,6 +520,8 @@ def from_uri(cls, uri):
             "playlist",
         ):
             return cls(uri, LinkType(parts[0]), parts[1], None)
+        elif len(parts) == 2 and parts[0] == "your":
+            return cls(uri, LinkType(parts[0]))
         elif len(parts) == 3 and parts[0] == "user" and parts[2] == "starred":
             if parsed_uri.scheme == "spotify":
                 return cls(uri, LinkType.PLAYLIST, None, parts[1])

tests/test_lookup.py

@@ -192,3 +192,87 @@ def test_lookup_of_playlist_uri_when_not_logged_in(
         "Failed to lookup 'spotify:playlist:alice:foo': "
         "Playlist Web API lookup failed" in caplog.text
     )
+
+
+def test_lookup_of_yourtracks_uri(web_client_mock, web_track_mock, provider):
+    web_saved_track_mock = {"track": web_track_mock}
+    web_client_mock.get_all.return_value = [
+        {"items": [web_saved_track_mock, web_saved_track_mock]},
+        {"items": [web_saved_track_mock, web_saved_track_mock]},
+    ]
+
+    results = provider.lookup("spotify:your:tracks")
+
+    assert len(results) == 4
+    for track in results:
+        assert track.uri == "spotify:track:abc"
+        assert track.name == "ABC 123"
+        assert track.bitrate == 160
+
+
+def test_lookup_of_youralbums_uri(
+    session_mock,
+    web_client_mock,
+    web_album_mock,
+    sp_album_browser_mock,
+    provider,
+):
+    web_saved_album_mock = {"album": web_album_mock}
+    web_client_mock.get_all.return_value = [
+        {"items": [web_saved_album_mock, web_saved_album_mock]},
+        {"items": [web_saved_album_mock, web_saved_album_mock]},
+    ]
+
+    sp_album_mock = sp_album_browser_mock.album
+    session_mock.get_link.return_value = sp_album_mock.link
+
+    results = provider.lookup("spotify:your:albums")
+
+    get_link_call = mock.call("spotify:album:def")
+    assert session_mock.get_link.call_args_list == [
+        get_link_call,
+        get_link_call,
+        get_link_call,
+        get_link_call,
+    ]
+    assert sp_album_mock.link.as_album.call_count == 4
+
+    assert sp_album_mock.browse.call_count == 4
+    load_call = mock.call(10)
+    assert sp_album_browser_mock.load.call_args_list == [
+        load_call,
+        load_call,
+        load_call,
+        load_call,
+    ]
+
+    assert len(results) == 8
+    for track in results:
+        assert track.uri == "spotify:track:abc"
+        assert track.name == "ABC 123"
+        assert track.album.uri == "spotify:album:def"
+        assert track.bitrate == 160
+
+
+def test_lookup_of_your_uri_when_not_logged_in(web_client_mock, provider):
+    web_client_mock.user_id = None
+
+    results = provider.lookup("spotify:your:tracks")
+
+    assert len(results) == 0
+
+
+def test_lookup_of_unhandled_your_uri(provider):
+    results = provider.lookup("spotify:your:artists")
+
+    assert len(results) == 0
+
+
+def test_lookup_of_invalid_your_uri(provider, caplog):
+    results = provider.lookup("spotify:your:tracks:invalid")
+
+    assert len(results) == 0
+    assert (
+        "Failed to lookup 'spotify:your:tracks:invalid': Could not parse"
+        in caplog.text
+    )

tests/test_web.py

@@ -1106,6 +1106,7 @@ def test_logged_in(self, spotify_client, user_id, expected):
         ("spotify:track:bar", web.LinkType.TRACK, "bar"),
         ("spotify:artist:blah", web.LinkType.ARTIST, "blah"),
         ("spotify:album:stuff", web.LinkType.ALBUM, "stuff"),
+        ("spotify:your:albums", web.LinkType.YOUR, None),
     ],
 )
 def test_weblink_from_uri_spotify_uri(uri, type_, id_):