From: Jonathan Bangert Date: Sat, 29 Jul 2023 11:37:26 +0000 (+0200) Subject: Delete helpers.py since it unused X-Git-Url: https://git.kitaultman.com/?a=commitdiff_plain;h=d9dd505bbc9d2f0ecf357c6262494dd0ffef7c27;p=music-assistant-server.git Delete helpers.py since it unused --- diff --git a/music_assistant/server/providers/deezer/helpers.py b/music_assistant/server/providers/deezer/helpers.py deleted file mode 100644 index c858f76b..00000000 --- a/music_assistant/server/providers/deezer/helpers.py +++ /dev/null @@ -1,322 +0,0 @@ -"""Helper module for parsing the Deezer API. Also helper for getting audio streams. - -This helpers file is an async wrapper around the excellent deezer-python package. -While the deezer-python package does an excellent job at parsing the Deezer results, -it is unfortunately not async, which is required for Music Assistant to run smoothly. -This also nicely separates the parsing logic from the Deezer provider logic. - -CREDITS: -deezer-python: https://github.com/browniebroke/deezer-python by @browniebroke -""" - -import asyncio -from dataclasses import dataclass - -import deezer - - -@dataclass -class Credential: - """Class for storing credentials.""" - - def __init__(self, app_id: int, app_secret: str, access_token: str): - """Set the correct things.""" - self.app_id = app_id - self.app_secret = app_secret - self.access_token = access_token - - app_id: int - app_secret: str - access_token: str - - -class DeezerClient: - """Async wrapper of the deezer-python library.""" - - _client: deezer.Client - _creds: Credential - user: deezer.User - - def __init__(self, creds: Credential, client: deezer.Client): - """Initialize the client.""" - self._creds = creds - self._client = client - self.user = self._client.get_user() - - async def get_deezer_client(self, creds: Credential) -> deezer.Client: # type: ignore - """ - Return a deezer-python Client. - - If credentials are given the client is authorized. - If no credentials are given the deezer client is not authorized. - - :param creds: Credentials. If none are given client is not authorized, defaults to None - :type creds: credential, optional - """ - if not isinstance(creds, Credential): - raise TypeError("Creds must be of type credential") - - def _authorize(): - return deezer.Client( - app_id=creds.app_id, app_secret=creds.app_secret, access_token=creds.access_token - ) - - return await asyncio.to_thread(_authorize) - - async def get_artist(self, artist_id: int) -> deezer.Artist: - """Async wrapper of the deezer-python get_artist function.""" - - def _get_artist(): - artist = self._client.get_artist(artist_id=artist_id) - return artist - - return await asyncio.to_thread(_get_artist) - - async def get_album(self, album_id: int) -> deezer.Album: - """Async wrapper of the deezer-python get_album function.""" - - def _get_album(): - album = self._client.get_album(album_id=album_id) - return album - - return await asyncio.to_thread(_get_album) - - async def get_playlist(self, playlist_id) -> deezer.Playlist: - """Async wrapper of the deezer-python get_playlist function.""" - - def _get_playlist(): - playlist = self._client.get_playlist(playlist_id=playlist_id) - return playlist - - return await asyncio.to_thread(_get_playlist) - - async def get_track(self, track_id: int) -> deezer.Track: - """Async wrapper of the deezer-python get_track function.""" - - def _get_track(): - track = self._client.get_track(track_id=track_id) - return track - - return await asyncio.to_thread(_get_track) - - async def get_user_artists(self) -> deezer.PaginatedList: - """Async wrapper of the deezer-python get_user_artists function.""" - - def _get_artist(): - artists = self._client.get_user_artists() - return artists - - return await asyncio.to_thread(_get_artist) - - async def get_user_playlists(self) -> deezer.PaginatedList: - """Async wrapper of the deezer-python get_user_playlists function.""" - - def _get_playlist(): - playlists = self._client.get_user().get_playlists() - return playlists - - return await asyncio.to_thread(_get_playlist) - - async def get_user_albums(self) -> deezer.PaginatedList: - """Async wrapper of the deezer-python get_user_albums function.""" - - def _get_album(): - albums = self._client.get_user_albums() - return albums - - return await asyncio.to_thread(_get_album) - - async def get_user_tracks(self) -> deezer.PaginatedList: - """Async wrapper of the deezer-python get_user_tracks function.""" - - def _get_track(): - tracks = self._client.get_user_tracks() - return tracks - - return await asyncio.to_thread(_get_track) - - async def add_user_albums(self, album_id: int) -> bool: - """Async wrapper of the deezer-python add_user_albums function.""" - - def _get_track(): - success = self._client.add_user_album(album_id=album_id) - return success - - return await asyncio.to_thread(_get_track) - - async def remove_user_albums(self, album_id: int) -> bool: - """Async wrapper of the deezer-python remove_user_albums function.""" - - def _get_track(): - success = self._client.remove_user_album(album_id=album_id) - return success - - return await asyncio.to_thread(_get_track) - - async def add_user_tracks(self, track_id: int) -> bool: - """Async wrapper of the deezer-python add_user_tracks function.""" - - def _get_track(): - success = self._client.add_user_track(track_id=track_id) - return success - - return await asyncio.to_thread(_get_track) - - async def remove_user_tracks(self, track_id: int) -> bool: - """Async wrapper of the deezer-python remove_user_tracks function.""" - - def _get_track(): - success = self._client.remove_user_track(track_id=track_id) - return success - - return await asyncio.to_thread(_get_track) - - async def add_user_artists(self, artist_id: int) -> bool: - """Async wrapper of the deezer-python add_user_artists function.""" - - def _get_artist(): - success = self._client.add_user_artist(artist_id=artist_id) - return success - - return await asyncio.to_thread(_get_artist) - - async def remove_user_artists(self, artist_id: int) -> bool: - """Async wrapper of the deezer-python remove_user_artists function.""" - - def _get_artist(): - success = self._client.remove_user_artist(artist_id=artist_id) - return success - - return await asyncio.to_thread(_get_artist) - - async def add_user_playlists(self, playlist_id: int) -> bool: - """Async wrapper of the deezer-python add_user_playlists function.""" - - def _get_playlist(): - success = self._client.add_user_playlist(playlist_id=playlist_id) - return success - - return await asyncio.to_thread(_get_playlist) - - async def remove_user_playlists(self, playlist_id: int) -> bool: - """Async wrapper of the deezer-python remove_user_playlists function.""" - - def _get_playlist(): - success = self._client.remove_user_playlist(playlist_id=playlist_id) - return success - - return await asyncio.to_thread(_get_playlist) - - async def search_album(self, query: str, limit: int = 5) -> list[deezer.Album]: - """Async wrapper of the deezer-python search_albums function.""" - - def _search(): - result = self._client.search_albums(query=query)[:limit] - return result - - return await asyncio.to_thread(_search) - - async def search_track(self, query: str, limit: int = 5) -> list[deezer.Track]: - """Async wrapper of the deezer-python search function.""" - - def _search(): - result = self._client.search(query=query)[:limit] - return result - - return await asyncio.to_thread(_search) - - async def search_artist(self, query: str, limit: int = 5) -> list[deezer.Artist]: - """Async wrapper of the deezer-python search_artist function.""" - - def _search(): - result = self._client.search_artists(query=query)[:limit] - return result - - return await asyncio.to_thread(_search) - - async def search_playlist(self, query: str, limit: int = 5) -> list[deezer.Playlist]: - """Async wrapper of the deezer-python search_playlist function.""" - - def _search(): - result = self._client.search_playlists(query=query)[:limit] - return result - - return await asyncio.to_thread(_search) - - async def get_album_from_track(self, track: deezer.Track) -> deezer.Album: - """Get track's artist.""" - - def _get_album_from_track(): - try: - return track.get_album() - except deezer.exceptions.DeezerErrorResponse: - return None - - return await asyncio.to_thread(_get_album_from_track) - - async def get_artist_from_track(self, track: deezer.Track) -> deezer.Artist: - """Get track's artist.""" - - def _get_artist_from_track(): - return track.get_artist() - - return await asyncio.to_thread(_get_artist_from_track) - - async def get_artist_from_album(self, album: deezer.Album) -> deezer.Artist: - """Get track's artist.""" - - def _get_artist_from_album(): - return album.get_artist() - - return await asyncio.to_thread(_get_artist_from_album) - - async def get_albums_by_artist(self, artist: deezer.Artist) -> deezer.PaginatedList: - """Get albums by an artist.""" - - def _get_albums_by_artist(): - return artist.get_albums() - - return await asyncio.to_thread(_get_albums_by_artist) - - async def get_artist_top(self, artist: deezer.Artist, limit: int = 25) -> deezer.PaginatedList: - """Get top tracks by an artist.""" - - def _get_artist_top(): - return artist.get_top()[:limit] - - return await asyncio.to_thread(_get_artist_top) - - async def get_recommended_tracks(self) -> deezer.PaginatedList: - """Get recommended tracks for user.""" - - def _get_recommended_tracks(): - return self._client.get_user_recommended_tracks() - - return await asyncio.to_thread(_get_recommended_tracks) - - async def create_playlist(self, playlist_name) -> deezer.Playlist: - """Create a playlist on deezer.""" - - def _create_playlist(): - playlist_id = self._client.create_playlist(playlist_name=playlist_name) - return self._client.get_playlist(playlist_id=playlist_id) - - return await asyncio.to_thread(_create_playlist) - - async def add_playlist_tracks(self, playlist_id: int, tracks: list[int]): - """Add tracks to playlist.""" - - def _add_playlist_tracks(): - playlist = self._client.get_playlist(playlist_id=playlist_id) - playlist.add_tracks(tracks=tracks) - - return await asyncio.to_thread(_add_playlist_tracks) - - async def remove_playlist_tracks(self, playlist_id: int, tracks: list[int]): - """Remove tracks from playlist.""" - - def _remove_playlist_tracks(): - playlist = self._client.get_playlist(playlist_id=playlist_id) - playlist.delete_tracks(tracks=tracks) - - return await asyncio.to_thread(_remove_playlist_tracks)