Typing: Tidal (music-assistant#1525)

* Typing: Tidal

* Update for changes in past month

music_assistant/server/helpers/throttle_retry.py

@@ -99,11 +99,11 @@ async def bypass(self) -> AsyncGenerator[None, None]:
 
 def throttle_with_retries(
     func: Callable[Concatenate[_ProviderT, _P], Awaitable[_R]],
-) -> Callable[Concatenate[_ProviderT, _P], Coroutine[Any, Any, _R | None]]:
+) -> Callable[Concatenate[_ProviderT, _P], Coroutine[Any, Any, _R]]:
     """Call async function using the throttler with retries."""
 
     @functools.wraps(func)
-    async def wrapper(self: _ProviderT, *args: _P.args, **kwargs: _P.kwargs) -> _R | None:
+    async def wrapper(self: _ProviderT, *args: _P.args, **kwargs: _P.kwargs) -> _R:
         """Call async function using the throttler with retries."""
         # the trottler attribute must be present on the class
         throttler: ThrottlerManager = self.throttler

music_assistant/server/providers/tidal/__init__.py

@@ -5,10 +5,11 @@
 import asyncio
 import base64
 import pickle
+from collections.abc import Callable
 from contextlib import suppress
 from datetime import datetime, timedelta
 from enum import StrEnum
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, ParamSpec, TypeVar, cast
 
 from tidalapi import Album as TidalAlbum
 from tidalapi import Artist as TidalArtist
@@ -46,6 +47,7 @@
     ProviderMapping,
     SearchResults,
     Track,
+    UniqueList,
 )
 from music_assistant.common.models.streamdetails import StreamDetails
 from music_assistant.server.helpers.auth import AuthenticationHelper
@@ -77,7 +79,7 @@
 )
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncGenerator, Awaitable, Callable
+    from collections.abc import AsyncGenerator, Awaitable
 
     from tidalapi.media import Lyrics as TidalLyrics
     from tidalapi.media import Stream as TidalStream
@@ -113,6 +115,9 @@
 BROWSE_URL = "https://tidal.com/browse"
 RESOURCES_URL = "https://resources.tidal.com/images"
 
+_R = TypeVar("_R")
+_P = ParamSpec("_P")
+
 
 class TidalQualityEnum(StrEnum):
     """Enum for Tidal Quality."""
@@ -171,10 +176,12 @@ async def get_config_entries(
     action: [optional] action key called from config entries UI.
     values: the (intermediate) raw values for config entries sent with the action.
     """
+    assert values is not None
+
     if action == CONF_ACTION_START_PKCE_LOGIN:
         async with AuthenticationHelper(mass, cast(str, values["session_id"])) as auth_helper:
-            quality: str = values.get(CONF_QUALITY) if values else None
-            base64_session = await tidal_auth_url(auth_helper, cast(str, quality))
+            quality = str(values.get(CONF_QUALITY))
+            base64_session = await tidal_auth_url(auth_helper, quality)
             values[CONF_TEMP_SESSION] = base64_session
             # Tidal is (ab)using the AuthenticationHelper just to send the user to an URL
             # there is no actual oauth callback happening, instead the user is redirected
@@ -183,9 +190,9 @@ async def get_config_entries(
             await asyncio.sleep(15)
 
     if action == CONF_ACTION_COMPLETE_PKCE_LOGIN:
-        quality: str = values.get(CONF_QUALITY) if values else None
-        pkce_url: str = values.get(CONF_OOPS_URL) if values else None
-        base64_session = values.get(CONF_TEMP_SESSION) if values else None
+        quality = str(values.get(CONF_QUALITY))
+        pkce_url = str(values.get(CONF_OOPS_URL))
+        base64_session = str(values.get(CONF_TEMP_SESSION))
         tidal_session = await tidal_pkce_login(base64_session, pkce_url)
         if not tidal_session.check_login():
             msg = "Authentication to Tidal failed"
@@ -201,7 +208,7 @@ async def get_config_entries(
         values[CONF_AUTH_TOKEN] = None
 
     if values.get(CONF_AUTH_TOKEN):
-        auth_entries = (
+        auth_entries: tuple[ConfigEntry, ...] = (
             ConfigEntry(
                 key="label_ok",
                 type=ConfigEntryType.LABEL,
@@ -347,14 +354,14 @@ class TidalProvider(MusicProvider):
     """Implementation of a Tidal MusicProvider."""
 
     _tidal_session: TidalSession | None = None
-    _tidal_user_id: str | None = None
+    _tidal_user_id: str
     # rate limiter needs to be specified on provider-level,
     # so make it an instance attribute
     throttler = ThrottlerManager(rate_limit=1, period=2)
 
     async def handle_async_init(self) -> None:
         """Handle async initialization of the provider."""
-        self._tidal_user_id: str = self.config.get_value(CONF_USER_ID)
+        self._tidal_user_id = str(self.config.get_value(CONF_USER_ID))
         try:
             self._tidal_session = await self._get_tidal_session()
         except Exception as err:
@@ -415,17 +422,15 @@ async def search(
         results = await search(tidal_session, search_query, media_types, limit)
 
         if results["artists"]:
-            for artist in results["artists"]:
-                parsed_results.artists.append(self._parse_artist(artist))
+            parsed_results.artists = [self._parse_artist(artist) for artist in results["artists"]]
         if results["albums"]:
-            for album in results["albums"]:
-                parsed_results.albums.append(self._parse_album(album))
+            parsed_results.albums = [self._parse_album(album) for album in results["albums"]]
         if results["playlists"]:
-            for playlist in results["playlists"]:
-                parsed_results.playlists.append(self._parse_playlist(playlist))
+            parsed_results.playlists = [
+                self._parse_playlist(playlist) for playlist in results["playlists"]
+            ]
         if results["tracks"]:
-            for track in results["tracks"]:
-                parsed_results.tracks.append(self._parse_track(track))
+            parsed_results.tracks = [self._parse_track(track) for track in results["tracks"]]
         return parsed_results
 
     async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
@@ -622,7 +627,7 @@ async def get_track(self, prov_track_id: str) -> Track:
             track = self._parse_track(track_obj)
             # get some extra details for the full track info
             with suppress(tidal_exceptions.MetadataNotAvailable, AttributeError):
-                lyrics: TidalLyrics = await asyncio.to_thread(track.lyrics)
+                lyrics: TidalLyrics = await asyncio.to_thread(track_obj.lyrics)
                 track.metadata.lyrics = lyrics.text
             return track
         except tidal_exceptions.ObjectNotFound as err:
@@ -655,7 +660,7 @@ async def _get_tidal_session(self) -> TidalSession:
             return self._tidal_session
         self._tidal_session = await self._load_tidal_session(
             token_type="Bearer",
-            quality=self.config.get_value(CONF_QUALITY),
+            quality=str(self.config.get_value(CONF_QUALITY)),
             access_token=str(self.config.get_value(CONF_AUTH_TOKEN)),
             refresh_token=str(self.config.get_value(CONF_REFRESH_TOKEN)),
             expiry_time=datetime.fromisoformat(str(self.config.get_value(CONF_EXPIRY_TIME))),
@@ -727,14 +732,16 @@ def _parse_artist(self, artist_obj: TidalArtist) -> Artist:
         if artist_obj.picture:
             picture_id = artist_obj.picture.replace("-", "/")
             image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
-            artist.metadata.images = [
-                MediaItemImage(
-                    type=ImageType.THUMB,
-                    path=image_url,
-                    provider=self.lookup_key,
-                    remotely_accessible=True,
-                )
-            ]
+            artist.metadata.images = UniqueList(
+                [
+                    MediaItemImage(
+                        type=ImageType.THUMB,
+                        path=image_url,
+                        provider=self.lookup_key,
+                        remotely_accessible=True,
+                    )
+                ]
+            )
 
         return artist
 
@@ -786,14 +793,16 @@ def _parse_album(self, album_obj: TidalAlbum) -> Album:
         if album_obj.cover:
             picture_id = album_obj.cover.replace("-", "/")
             image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
-            album.metadata.images = [
-                MediaItemImage(
-                    type=ImageType.THUMB,
-                    path=image_url,
-                    provider=self.lookup_key,
-                    remotely_accessible=True,
-                )
-            ]
+            album.metadata.images = UniqueList(
+                [
+                    MediaItemImage(
+                        type=ImageType.THUMB,
+                        path=image_url,
+                        provider=self.lookup_key,
+                        remotely_accessible=True,
+                    )
+                ]
+            )
 
         return album
 
@@ -828,7 +837,7 @@ def _parse_track(
         )
         if track_obj.isrc:
             track.external_ids.add((ExternalID.ISRC, track_obj.isrc))
-        track.artists = []
+        track.artists = UniqueList()
         for track_artist in track_obj.artists:
             artist = self._parse_artist(track_artist)
             track.artists.append(artist)
@@ -847,14 +856,16 @@ def _parse_track(
             if track_obj.album.cover:
                 picture_id = track_obj.album.cover.replace("-", "/")
                 image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
-                track.metadata.images = [
-                    MediaItemImage(
-                        type=ImageType.THUMB,
-                        path=image_url,
-                        provider=self.lookup_key,
-                        remotely_accessible=True,
-                    )
-                ]
+                track.metadata.images = UniqueList(
+                    [
+                        MediaItemImage(
+                            type=ImageType.THUMB,
+                            path=image_url,
+                            provider=self.lookup_key,
+                            remotely_accessible=True,
+                        )
+                    ]
+                )
         return track
 
     def _parse_playlist(self, playlist_obj: TidalPlaylist) -> Playlist:
@@ -884,27 +895,32 @@ def _parse_playlist(self, playlist_obj: TidalPlaylist) -> Playlist:
         if picture := (playlist_obj.square_picture or playlist_obj.picture):
             picture_id = picture.replace("-", "/")
             image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
-            playlist.metadata.images = [
-                MediaItemImage(
-                    type=ImageType.THUMB,
-                    path=image_url,
-                    provider=self.lookup_key,
-                    remotely_accessible=True,
-                )
-            ]
+            playlist.metadata.images = UniqueList(
+                [
+                    MediaItemImage(
+                        type=ImageType.THUMB,
+                        path=image_url,
+                        provider=self.lookup_key,
+                        remotely_accessible=True,
+                    )
+                ]
+            )
 
         return playlist
 
     async def _iter_items(
-        self, func: Awaitable | Callable, *args, **kwargs
-    ) -> AsyncGenerator[Any, None]:
+        self,
+        func: Callable[_P, list[_R]] | Callable[_P, Awaitable[list[_R]]],
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> AsyncGenerator[_R, None]:
         """Yield all items from a larger listing."""
         offset = 0
         while True:
             if asyncio.iscoroutinefunction(func):
-                chunk = await func(*args, **kwargs, offset=offset)
+                chunk = await func(*args, **kwargs, offset=offset)  # type: ignore[arg-type]
             else:
-                chunk = await asyncio.to_thread(func, *args, **kwargs, offset=offset)
+                chunk = await asyncio.to_thread(func, *args, **kwargs, offset=offset)  # type: ignore[arg-type]
             offset += len(chunk)
             for item in chunk:
                 yield item

music_assistant/server/providers/tidal/helpers.py

@@ -50,7 +50,7 @@ async def library_items_add_remove(
     item_id: str,
     media_type: MediaType,
     add: bool = True,
-) -> None:
+) -> bool:
     """Async wrapper around the tidalapi Favorites.items add/remove function."""
 
     def inner() -> bool:
@@ -112,7 +112,7 @@ def inner() -> list[TidalAlbum]:
             msg = "Tidal API rate limit reached"
             raise ResourceTemporarilyUnavailable(msg)
         else:
-            all_albums = artist_obj.get_albums(limit=DEFAULT_LIMIT)
+            all_albums: list[TidalAlbum] = artist_obj.get_albums(limit=DEFAULT_LIMIT)
             # extend with EPs and singles
             all_albums.extend(artist_obj.get_ep_singles(limit=DEFAULT_LIMIT))
             # extend with compilations
@@ -189,7 +189,7 @@ def inner() -> TidalTrack:
 async def get_stream(track: TidalTrack) -> TidalStream:
     """Async wrapper around the tidalapi Track.get_stream_url function."""
 
-    def inner() -> str:
+    def inner() -> TidalStream:
         try:
             return track.get_stream()
         except ObjectNotFound as err:

mypy.ini

@@ -21,4 +21,4 @@ disallow_untyped_decorators = true
 disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
-packages=tests,music_assistant.client,music_assistant.common,music_assistant.server.providers.builtin,music_assistant.server.providers.filesystem_local,music_assistant.server.providers.filesystem_smb,music_assistant.server.providers.fully_kiosk,music_assistant.server.providers.jellyfin,music_assistant.server.providers.plex,music_assistant.server.providers.radiobrowser,music_assistant.server.providers.test,music_assistant.server.providers.theaudiodb
+packages=tests,music_assistant.client,music_assistant.common,music_assistant.server.providers.builtin,music_assistant.server.providers.filesystem_local,music_assistant.server.providers.filesystem_smb,music_assistant.server.providers.fully_kiosk,music_assistant.server.providers.jellyfin,music_assistant.server.providers.plex,music_assistant.server.providers.radiobrowser,music_assistant.server.providers.test,music_assistant.server.providers.theaudiodb,music_assistant.server.providers.tidal