gw_client: GWClient
creds: Credential
_throttler: Throttler
-
- @property
- def name(self) -> str:
- """Return (custom) friendly name for this provider instance."""
- if client := getattr(self, "client", None):
- return f"Deezer - {client.user.name}"
- return super().name
+ user: deezer.resources.User
async def handle_setup(self) -> None:
"""Set up the Deezer provider."""
except Exception as error:
raise LoginFailed("Invalid login credentials") from error
+ self.user = await self.client.get_user()
+
self.gw_client = GWClient(self.mass.http_session, self.config.get_value(CONF_ACCESS_TOKEN))
await self.gw_client.setup()
async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
"""Retrieve all library artists from Deezer."""
- for artist in await self.client.get_user_artists():
- yield self.parse_artist(artist=artist)
+ async for artist in await self.client.get_user_artists():
+ yield await self.parse_artist(artist=artist)
async def get_library_albums(self) -> AsyncGenerator[Album, None]:
"""Retrieve all library albums from Deezer."""
- for album in await self.client.get_user_albums():
- yield self.parse_album(album=album)
+ async for album in await self.client.get_user_albums():
+ yield await self.parse_album(album=album)
async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
"""Retrieve all library playlists from Deezer."""
- for playlist in await self.client.get_user_playlists():
- yield self.parse_playlist(playlist=playlist)
+ async for playlist in await self.client.get_user_playlists():
+ yield await self.parse_playlist(playlist=playlist)
async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
"""Retrieve all library tracks from Deezer."""
- for track in await self.client.get_user_tracks():
- yield self.parse_track(track=track, user_country=self.gw_client.user_country)
+ async for track in await self.client.get_user_tracks():
+ yield await self.parse_track(track=track, user_country=self.gw_client.user_country)
async def get_artist(self, prov_artist_id: str) -> Artist:
"""Get full artist details by id."""
- return self.parse_artist(artist=await self.client.get_artist(artist_id=int(prov_artist_id)))
+ return await self.parse_artist(
+ artist=await self.client.get_artist(artist_id=int(prov_artist_id))
+ )
async def get_album(self, prov_album_id: str) -> Album:
"""Get full album details by id."""
try:
- return self.parse_album(album=await self.client.get_album(album_id=int(prov_album_id)))
+ return await self.parse_album(
+ album=await self.client.get_album(album_id=int(prov_album_id))
+ )
except deezer.exceptions.DeezerErrorResponse as error:
self.logger.warning("Failed getting album: %s", error)
return Album(prov_album_id, self.instance_id, "Not Found")
async def get_playlist(self, prov_playlist_id: str) -> Playlist:
"""Get full playlist details by id."""
- return self.parse_playlist(
+ return await self.parse_playlist(
playlist=await self.client.get_playlist(playlist_id=int(prov_playlist_id)),
)
async def get_track(self, prov_track_id: str) -> Track:
"""Get full track details by id."""
- return self.parse_track(
+ return await self.parse_track(
track=await self.client.get_track(track_id=int(prov_track_id)),
user_country=self.gw_client.user_country,
)
"""Get all albums in a playlist."""
album = await self.client.get_album(album_id=int(prov_album_id))
return [
- self.parse_track(track=track, user_country=self.gw_client.user_country)
+ await self.parse_track(track=track, user_country=self.gw_client.user_country)
for track in album.tracks
]
"""Get all tracks in a playlist."""
playlist = await self.client.get_playlist(playlist_id=prov_playlist_id)
for count, track in enumerate(playlist.tracks, start=1):
- track_parsed = self.parse_track(track=track, user_country=self.gw_client.user_country)
+ track_parsed = await self.parse_track(
+ track=track, user_country=self.gw_client.user_country
+ )
track_parsed.position = count
track_parsed.id = track.id
yield track_parsed
artist = await self.client.get_artist(artist_id=int(prov_artist_id))
albums = []
for album in await self.client.get_albums_by_artist(artist=artist):
- albums.append(self.parse_album(album=album))
+ albums.append(await self.parse_album(album=album))
return albums
async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
artist = await self.client.get_artist(artist_id=int(prov_artist_id))
top_tracks = await self.client.get_artist_top(artist=artist, limit=25)
return [
- self.parse_track(track=track, user_country=self.gw_client.user_country)
+ await self.parse_track(track=track, user_country=self.gw_client.user_country)
for track in top_tracks
]
name="Recommendations",
label="recommendations",
items=[
- self.parse_track(track=track, user_country=self.gw_client.user_country)
+ await self.parse_track(track=track, user_country=self.gw_client.user_country)
for track in await self.client.get_recommended_tracks()
],
)
async def create_playlist(self, name: str) -> Playlist:
"""Create a new playlist on provider with given name."""
playlist = await self.client.create_playlist(playlist_name=name)
- return self.parse_playlist(playlist=playlist)
+ return await self.parse_playlist(playlist=playlist)
async def get_stream_details(self, item_id: str) -> StreamDetails | None:
"""Return the content details for the given track when it will be streamed."""
### PARSING METADATA FUNCTIONS ###
- def parse_metadata_track(self, track: deezer.Track) -> MediaItemMetadata:
+ async def parse_metadata_track(self, track: deezer.Track) -> MediaItemMetadata:
"""Parse the track metadata."""
try:
return MediaItemMetadata(
preview=track.preview,
)
- def parse_metadata_album(self, album: deezer.Album) -> MediaItemMetadata:
+ async def parse_metadata_album(self, album: deezer.Album) -> MediaItemMetadata:
"""Parse the album metadata."""
return MediaItemMetadata(
images=[MediaItemImage(type=ImageType.THUMB, path=album.cover_big)],
)
- def parse_metadata_artist(self, artist: deezer.Artist) -> MediaItemMetadata:
+ async def parse_metadata_artist(self, artist: deezer.Artist) -> MediaItemMetadata:
"""Parse the artist metadata."""
return MediaItemMetadata(
images=[MediaItemImage(type=ImageType.THUMB, path=artist.picture_big)],
)
### PARSING FUNCTIONS ###
- def parse_artist(self, artist: deezer.Artist) -> Artist:
+ async def parse_artist(self, artist: deezer.Artist) -> Artist:
"""Parse the deezer-python artist to a MASS artist."""
return Artist(
item_id=str(artist.id),
metadata=self.parse_metadata_artist(artist=artist),
)
- def parse_album(self, album: deezer.Album) -> Album:
+ async def parse_album(self, album: deezer.Album) -> Album:
"""Parse the deezer-python album to a MASS album."""
return Album(
album_type=AlbumType(album.type),
metadata=self.parse_metadata_album(album=album),
)
- def parse_playlist(self, playlist: deezer.Playlist) -> Playlist:
+ async def parse_playlist(self, playlist: deezer.Playlist) -> Playlist:
"""Parse the deezer-python playlist to a MASS playlist."""
return Playlist(
item_id=str(playlist.id),
is_editable=playlist.creator.id == self.client.user.id,
)
- def parse_track(self, track: deezer.Track, user_country: str) -> Track:
+ async def parse_track(self, track: deezer.Track, user_country: str) -> Track:
"""Parse the deezer-python track to a MASS track."""
return Track(
item_id=str(track.id),
) -> list[Track]:
"""Search for tracks and parse them."""
deezer_tracks = await self.client.search_track(query=query, limit=limit)
- return [self.parse_track(track, user_country) for track in deezer_tracks]
+ return [await self.parse_track(track, user_country) for track in deezer_tracks]
async def search_and_parse_artists(self, query: str, limit: int = 5) -> list[Artist]:
"""Search for artists and parse them."""
deezer_artist = await self.client.search_artist(query=query, limit=limit)
- return [self.parse_artist(artist=artist) for artist in deezer_artist]
+ return [await self.parse_artist(artist=artist) for artist in deezer_artist]
async def search_and_parse_albums(self, query: str, limit: int = 5) -> list[Album]:
"""Search for album and parse them."""
deezer_albums = await self.client.search_album(query=query, limit=limit)
- return [self.parse_album(album=album) for album in deezer_albums]
+ return [await self.parse_album(album=album) for album in deezer_albums]
async def search_and_parse_playlists(self, query: str, limit: int = 5) -> list[Playlist]:
"""Search for playlists and parse them."""
deezer_playlists = await self.client.search_playlist(query=query, limit=limit)
- return [self.parse_playlist(playlist=playlist) for playlist in deezer_playlists]
+ return [await self.parse_playlist(playlist=playlist) for playlist in deezer_playlists]
### OTHER PARSING FUNCTIONS ###
- def _get_album(self, track: deezer.Track) -> Album | None:
+ async def _get_album(self, track: deezer.Track) -> Album | None:
try:
- return self.parse_album(album=track.get_album())
+ return await self.parse_album(album=track.get_album())
except AttributeError:
return None
projects / music-assistant-server.git / commitdiff