"""Async wrapper around the tidalapi Session function."""
def inner() -> TidalSession:
- config = TidalConfig(quality=TidalQuality[quality], item_limit=10000, alac=False)
+ config = TidalConfig(quality=quality, item_limit=10000, alac=False)
session = TidalSession(config=config)
login, future = session.login_oauth()
auth_helper.send_url(f"https://{login.verification_uri_complete}")
required=True,
description="The Tidal Quality you wish to use",
options=(
+ ConfigValueOption(title=TidalQuality.low_96k, value=TidalQuality.low_96k),
+ ConfigValueOption(title=TidalQuality.low_320k, value=TidalQuality.low_320k),
ConfigValueOption(
- title=TidalQuality.low_96k.value, value=TidalQuality.low_96k.name
+ title=TidalQuality.high_lossless,
+ value=TidalQuality.high_lossless,
),
- ConfigValueOption(
- title=TidalQuality.low_320k.value, value=TidalQuality.low_320k.name
- ),
- ConfigValueOption(
- title=TidalQuality.high_lossless.value,
- value=TidalQuality.high_lossless.name,
- ),
- ConfigValueOption(title=TidalQuality.hi_res.value, value=TidalQuality.hi_res.name),
+ ConfigValueOption(title=TidalQuality.hi_res, value=TidalQuality.hi_res),
),
- default_value=TidalQuality.high_lossless.name,
+ default_value=TidalQuality.high_lossless,
value=values.get(CONF_QUALITY) if values else None,
),
ConfigEntry(
async def _load_tidal_session(
self,
token_type: str,
- quality: TidalQuality,
+ quality: str,
access_token: str,
refresh_token: str,
expiry_time: datetime | None = None,
"""Load the tidalapi Session."""
def inner() -> TidalSession:
- config = TidalConfig(quality=TidalQuality[quality], item_limit=10000, alac=False)
+ config = TidalConfig(quality=quality, item_limit=10000, alac=False)
session = TidalSession(config=config)
session.load_oauth_session(token_type, access_token, refresh_token, expiry_time)
return session
def _is_hi_res(self, track_obj: TidalTrack) -> bool:
"""Check if track is hi-res."""
- hi_res: bool = track_obj.audio_quality.value == "HI_RES"
+ hi_res: bool = track_obj.audio_quality == "HI_RES"
return hi_res
import asyncio
import logging
-from requests import HTTPError
from tidalapi import Album as TidalAlbum
from tidalapi import Artist as TidalArtist
from tidalapi import Favorites as TidalFavorites
from tidalapi import Session as TidalSession
from tidalapi import Track as TidalTrack
from tidalapi import UserPlaylist as TidalUserPlaylist
+from tidalapi.exceptions import ObjectNotFound, TooManyRequests
from music_assistant.common.models.enums import MediaType
from music_assistant.common.models.errors import MediaNotFoundError
def inner() -> TidalArtist:
try:
return TidalArtist(session, prov_artist_id)
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Artist {prov_artist_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Artist {prov_artist_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
all_albums.extend(albums)
all_albums.extend(eps_singles)
all_albums.extend(compilations)
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Artist {prov_artist_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Artist {prov_artist_id} not found"
+ raise MediaNotFoundError(msg) from err
else:
return all_albums
def inner() -> TidalAlbum:
try:
return TidalAlbum(session, prov_album_id)
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Album {prov_album_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Album {prov_album_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
def inner() -> TidalTrack:
try:
return TidalTrack(session, prov_track_id)
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Track {prov_track_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Track {prov_track_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
try:
track_url: str = TidalTrack(session, prov_track_id).get_url()
return track_url
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Track {prov_track_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Track {prov_track_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
limit=DEFAULT_LIMIT
)
return tracks
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Album {prov_album_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Album {prov_album_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
def inner() -> TidalPlaylist:
try:
return TidalPlaylist(session, prov_playlist_id)
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Playlist {prov_playlist_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Playlist {prov_playlist_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
limit=limit, offset=offset
)
return tracks
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Playlist {prov_playlist_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Playlist {prov_playlist_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)
limit=limit
)
return tracks
- except HTTPError as err:
- if err.response.status_code == 404:
- msg = f"Track {prov_track_id} not found"
- raise MediaNotFoundError(msg) from err
- raise
+ except (ObjectNotFound, TooManyRequests) as err:
+ msg = f"Track {prov_track_id} not found"
+ raise MediaNotFoundError(msg) from err
return await asyncio.to_thread(inner)