Add method to check if a single URI is added to the library (#543)

Author: joostlek

src/spotifyaio/spotify.py

@@ -253,6 +253,11 @@ async def are_albums_saved(self, album_ids: list[str]) -> dict[str, bool]:
         body: list[bool] = orjson.loads(response)  # pylint: disable=no-member
         return dict(zip(identifiers, body))
 
+    async def is_album_saved(self, album_id: str) -> bool:
+        """Check if album is saved."""
+        identifier = get_identifier(album_id)
+        return (await self.are_albums_saved([identifier]))[identifier]
+
     async def get_new_releases(self) -> list[SimplifiedAlbum]:
         """Get new releases."""
         params: dict[str, Any] = {"limit": 48}
@@ -450,6 +455,24 @@ async def are_episodes_saved(self, episode_ids: list[str]) -> dict[str, bool]:
         body: list[bool] = orjson.loads(response)  # pylint: disable=no-member
         return dict(zip(identifiers, body))
 
+    async def is_episode_saved(self, episode_id: str) -> bool:
+        """Check if episode is saved."""
+        identifier = get_identifier(episode_id)
+        return (await self.are_episodes_saved([identifier]))[identifier]
+
+    async def is_added_to_library(self, uri: str) -> bool:
+        """Check if item is added to library."""
+        if uri.startswith("spotify:track:"):
+            return await self.is_track_saved(uri)
+        if uri.startswith("spotify:episode:"):
+            return await self.is_episode_saved(uri)
+        if uri.startswith("spotify:show:"):
+            return await self.is_show_saved(uri)
+        if uri.startswith("spotify:album:"):
+            return await self.is_album_saved(uri)
+        msg = "Invalid URI format"
+        raise ValueError(msg)
+
     async def get_playback(self) -> PlaybackState | None:
         """Get playback state."""
         response = await self._get(
@@ -784,6 +807,11 @@ async def are_shows_saved(self, show_ids: list[str]) -> dict[str, bool]:
         body: list[bool] = orjson.loads(response)  # pylint: disable=no-member
         return dict(zip(identifiers, body))
 
+    async def is_show_saved(self, show_id: str) -> bool:
+        """Check if show is saved."""
+        identifier = get_identifier(show_id)
+        return (await self.are_shows_saved([identifier]))[identifier]
+
     # Get a track
 
     # Get several tracks
@@ -831,6 +859,11 @@ async def are_tracks_saved(self, track_ids: list[str]) -> dict[str, bool]:
         body: list[bool] = orjson.loads(response)  # pylint: disable=no-member
         return dict(zip(identifiers, body))
 
+    async def is_track_saved(self, track_id: str) -> bool:
+        """Check if track is saved."""
+        identifier = get_identifier(track_id)
+        return (await self.are_tracks_saved([identifier]))[identifier]
+
     async def get_audio_features(self, track_id: str) -> AudioFeatures:
         """Get audio features."""
         identifier = get_identifier(track_id)

tests/test_spotify.py

@@ -1873,6 +1873,26 @@ async def test_check_saved_episodes(
     )
 
 
+async def test_check_saved_episode(
+    responses: aioresponses,
+    authenticated_client: SpotifyClient,
+) -> None:
+    """Test checking saved episode."""
+    responses.get(
+        f"{SPOTIFY_URL}/v1/me/episodes/contains?ids=18yVqkdbdRvS24c0Ilj2ci",
+        status=200,
+        body=load_fixture("tracks_saved.json"),
+    )
+    assert await authenticated_client.is_episode_saved("18yVqkdbdRvS24c0Ilj2ci") is True
+    responses.assert_called_once_with(
+        f"{SPOTIFY_URL}/v1/me/episodes/contains",
+        METH_GET,
+        headers=HEADERS,
+        params={"ids": "18yVqkdbdRvS24c0Ilj2ci"},
+        json=None,
+    )
+
+
 async def test_check_no_saved_episodes(
     responses: aioresponses,
     authenticated_client: SpotifyClient,
@@ -2470,6 +2490,72 @@ async def test_check_saved_tracks(
     )
 
 
+async def test_check_saved_track(
+    responses: aioresponses,
+    authenticated_client: SpotifyClient,
+) -> None:
+    """Test checking saved track."""
+    responses.get(
+        f"{SPOTIFY_URL}/v1/me/tracks/contains?ids=18yVqkdbdRvS24c0Ilj2ci",
+        status=200,
+        body=load_fixture("tracks_saved.json"),
+    )
+    assert await authenticated_client.is_track_saved("18yVqkdbdRvS24c0Ilj2ci") is True
+    responses.assert_called_once_with(
+        f"{SPOTIFY_URL}/v1/me/tracks/contains",
+        METH_GET,
+        headers=HEADERS,
+        params={"ids": "18yVqkdbdRvS24c0Ilj2ci"},
+        json=None,
+    )
+
+
+@pytest.mark.parametrize(
+    ("prefix", "url_part"),
+    [
+        ("spotify:track:", "tracks"),
+        ("spotify:episode:", "episodes"),
+        ("spotify:show:", "shows"),
+        ("spotify:album:", "albums"),
+    ],
+)
+async def test_check_saved_item(
+    responses: aioresponses,
+    authenticated_client: SpotifyClient,
+    prefix: str,
+    url_part: str,
+) -> None:
+    """Test checking saved track."""
+    responses.get(
+        f"{SPOTIFY_URL}/v1/me/{url_part}/contains?ids=18yVqkdbdRvS24c0Ilj2ci",
+        status=200,
+        body=load_fixture("tracks_saved.json"),
+    )
+    assert (
+        await authenticated_client.is_added_to_library(
+            f"{prefix}18yVqkdbdRvS24c0Ilj2ci"
+        )
+        is True
+    )
+    responses.assert_called_once_with(
+        f"{SPOTIFY_URL}/v1/me/{url_part}/contains",
+        METH_GET,
+        headers=HEADERS,
+        params={"ids": "18yVqkdbdRvS24c0Ilj2ci"},
+        json=None,
+    )
+
+
+async def test_checking_invalid_uri(
+    authenticated_client: SpotifyClient,
+) -> None:
+    """Test checking saved track."""
+    with pytest.raises(ValueError, match="Invalid URI format"):
+        await authenticated_client.is_added_to_library(
+            "spotify:new:18yVqkdbdRvS24c0Ilj2ci"
+        )
+
+
 async def test_check_no_saved_tracks(
     responses: aioresponses,
     authenticated_client: SpotifyClient,