diff --git a/endpoints/lastfm.py b/endpoints/lastfm.py index ad30c5d..65c4a46 100644 --- a/endpoints/lastfm.py +++ b/endpoints/lastfm.py @@ -10,7 +10,7 @@ from .constructors import ValidArtistSearchRequest, ValidAlbumDetailRequest,\ class LastFM(FastAPI): """Last.FM Endpoints""" - def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called + def __init__(self, app: FastAPI, util, constants, glob_state) -> None: # pylint: disable=super-init-not-called self.app = app self.util = util self.constants = constants diff --git a/lastfm_wrapper.py b/lastfm_wrapper.py index 129c4f7..1ff1577 100644 --- a/lastfm_wrapper.py +++ b/lastfm_wrapper.py @@ -3,7 +3,7 @@ import traceback import logging -from typing import Union +from typing import Optional import regex from aiohttp import ClientSession, ClientTimeout from constants import Constants @@ -12,180 +12,220 @@ from constants import Constants class LastFM: """LastFM Endpoints""" - def __init__(self, noInit: Union[None, bool] = False): # pylint: disable=unused-argument + def __init__(self, noInit: Optional[bool] = False) -> None: # pylint: disable=unused-argument self.creds = Constants().LFM_CREDS - self.api_base_url = "https://ws.audioscrobbler.com/2.0/?method=" + self.api_base_url: str = "https://ws.audioscrobbler.com/2.0/?method=" - async def search_artist(self, artist=None): + async def search_artist(self, artist: Optional[str] = None) -> dict: """Search LastFM for an artist""" try: if artist is None: return { - 'err': 'No artist specified.' + 'err': 'No artist specified.', } async with ClientSession() as session: async with await session.get(f"{self.api_base_url}artist.getinfo&artist={artist}&api_key={self.creds.get('key')}&autocorrect=1&format=json", timeout=ClientTimeout(connect=3, sock_read=8)) as request: - assert request.status in [200, 204] - data = await request.json() + request.raise_for_status() + data: dict = await request.json() data = data.get('artist') - - logging.debug("Using data:\n%s", data) - # return data.get('results') - retObj = { - 'id': data.get('mbid'), - 'touring': data.get('ontour'), - 'name': data.get('name'), - 'bio': data.get('bio').get('summary').strip().split(" dict: - """Get Track Info from LastFM""" + async def get_track_info(self, artist: Optional[str] = None, + track: Optional[str] = None) -> dict: + """ + Get Track Info from LastFM + Args: + artist (Optional[str]) + track (Optional[str]) + Returns: + dict + """ try: - if artist is None or track is None: + if not artist or not track: logging.info("inv request") return { - 'err': 'Invalid/No artist or track specified' + 'err': 'Invalid/No artist or track specified', } async with ClientSession() as session: async with await session.get(f"{self.api_base_url}track.getInfo&api_key={self.creds.get('key')}&autocorrect=1&artist={artist}&track={track}&format=json", timeout=ClientTimeout(connect=3, sock_read=8)) as request: - assert request.status in [200, 204] - data = await request.json() + request.raise_for_status() + data: dict = await request.json() data = data.get('track') - retObj = { + ret_obj: dict = { 'artist_mbid': data.get('artist').get('mbid'), - 'album': data.get('album').get('title') + 'album': data.get('album').get('title'), } - logging.debug("Returning:\n%s", retObj) - return retObj + return ret_obj except: traceback.print_exc() return { - 'err': 'General Failure' + 'err': 'General Failure', } - async def get_album_tracklist(self, artist=None, album=None) -> dict: - """Get Album Tracklist""" + async def get_album_tracklist(self, artist: Optional[str] = None, + album: Optional[str] = None) -> dict: + """ + Get Album Tracklist + Args: + artist (str) + album (str) + Returns: + dict + """ try: if artist is None or album is None: logging.info("inv request") return { - 'err': 'No artist or album specified' + 'err': 'No artist or album specified', } - tracks = await self.get_release(artist=artist, album=album) - tracks = tracks.get('tracks') - retObj = { - 'tracks': tracks + tracks: list|dict = await self.get_release(artist=artist, album=album) + tracks: list|dict = tracks.get('tracks') + ret_obj: dict = { + 'tracks': tracks, } - logging.debug("Returning:\n%s", retObj) - return retObj + return ret_obj except: traceback.print_exc() return { - 'err': 'General Failure' + 'err': 'General Failure', } - async def get_artist_albums(self, artist=None) -> dict|list[dict]: - """Get Artists Albums from LastFM""" + async def get_artist_albums(self, artist: Optional[str] = None) -> dict|list[dict]: + """ + Get Artists Albums from LastFM + Args: + artist (Optional[str]) + Returns: + dict|list[dict] + """ try: if artist is None: return { - 'err': 'No artist specified.' + 'err': 'No artist specified.', } async with ClientSession() as session: async with await session.get(f"{self.api_base_url}artist.gettopalbums&artist={artist}&api_key={self.creds.get('key')}&autocorrect=1&format=json", timeout=ClientTimeout(connect=3, sock_read=8)) as request: - assert request.status in [200, 204] - data = await request.json() - data = data.get('topalbums').get('album') - retObj = [ + request.raise_for_status() + data: dict = await request.json() + data: str = data.get('topalbums').get('album') + ret_obj: dict = [ { 'title': item.get('name') } for item in data if not(item.get('name').lower() == "(null)") and int(item.get('playcount')) >= 50 ] - return retObj + return ret_obj except: traceback.print_exc() return { - 'err': 'Failed' + 'err': 'Failed', } - async def get_artist_id(self, artist=None): - """Get Artist ID from LastFM""" + async def get_artist_id(self, artist: Optional[str] = None) -> int|dict: + """ + Get Artist ID from LastFM + Args: + artist (Optional[str]) + Returns: + int|dict + """ try: if artist is None: return { - 'err': 'No artist specified.' + 'err': 'No artist specified.', } - artist_search = await self.search_artist(artist=artist) - if artist_search is None or len(artist_search) < 1: + artist_search: dict = await self.search_artist(artist=artist) + if not artist_search: logging.debug("[get_artist_id] Throwing no result error") return { - 'err': 'No results.' + 'err': 'No results.', } - artist_id = artist_search[0].get('id') + artist_id: int = int(artist_search[0].get('id', 0)) return artist_id except: traceback.print_exc() return { - 'err': 'Failed' + 'err': 'Failed', } - async def get_artist_info_by_id(self, artist_id=None): - """Get Artist info by ID from LastFM""" + async def get_artist_info_by_id(self, artist_id: Optional[int] = None) -> dict: + """ + Get Artist info by ID from LastFM + Args: + artist_id (Optional[int]) + Returns: + dict + """ try: - if artist_id is None or not str(artist_id).isnumeric(): + if not artist_id or not str(artist_id).isnumeric(): return { - 'err': 'Invalid/no artist_id specified.' + 'err': 'Invalid/no artist_id specified.', } + + req_url: str = f"{self.api_base_url}artists/{artist_id}?key={self.creds.get('key')}\ + &secret={self.creds.get('secret')}" async with ClientSession() as session: - async with await session.get(f"{self.api_base_url}artists/{artist_id}?key={self.creds.get('key')}&secret={self.creds.get('secret')}", + async with await session.get(req_url, timeout=ClientTimeout(connect=3, sock_read=8)) as request: - assert request.status in [200, 204] - data = await request.json() + request.raise_for_status() + data: dict = await request.json() - retObj = { + ret_obj: dict = { 'id': data.get('id'), 'name': data.get('name'), 'profile': regex.sub(r"(\[(\/{0,})(u|b|i)])", "", data.get('profile')), - 'members': data.get('members') + 'members': data.get('members'), } - return retObj + return ret_obj except: traceback.print_exc() return { - 'err': 'Failed' + 'err': 'Failed', } - async def get_artist_info(self, artist=None): - """Get Artist Info from LastFM""" + async def get_artist_info(self, artist: Optional[str] = None) -> dict: + """ + Get Artist Info from LastFM + Args: + artist (Optional[str]) + Returns: + dict + """ try: - if artist is None: + if not artist: return { - 'err': 'No artist specified.' + 'err': 'No artist specified.', } - artist_id = await self.get_artist_id(artist=artist) - if artist_id is None: + artist_id: int = await self.get_artist_id(artist=artist) + if not artist_id: return { 'err': 'Failed', } - artist_info = await self.get_artist_info_by_id(artist_id=artist_id) - if artist_info is None: + artist_info: dict = await self.get_artist_info_by_id(artist_id=artist_id) + if not artist_info: return { 'err': 'Failed', } @@ -193,56 +233,64 @@ class LastFM: except: traceback.print_exc() return { - 'err': 'Failed' + 'err': 'Failed', } - async def get_release(self, artist=None, album=None): - """Get Release info from LastFM""" + async def get_release(self, artist: Optional[str] = None, + album: Optional[str] = None) -> dict: + """ + Get Release info from LastFM + Args: + artist (Optional[str]) + album (Optioanl[str]) + Returns: + dict + """ try: - if artist is None or album is None: + if not artist or not album: return { - 'err': 'Invalid artist/album pair' + 'err': 'Invalid artist/album pair', } + req_url: str = f"{self.api_base_url}album.getinfo&artist={artist}&album={album}\ + &api_key={self.creds.get('key')}&autocorrect=1&format=json" async with ClientSession() as session: - async with await session.get(f"{self.api_base_url}album.getinfo&artist={artist}&album={album}&api_key={self.creds.get('key')}&autocorrect=1&format=json", + async with await session.get(req_url, timeout=ClientTimeout(connect=3, sock_read=8)) as request: - assert request.status in [200, 204] - data = await request.json() - data = data.get('album') - retObj = { + request.raise_for_status() + data: dict = await request.json() + data: dict = data.get('album') + ret_obj: dict = { 'id': data.get('mbid'), 'artists': data.get('artist'), 'tags': data.get('tags'), 'title': data.get('name'), - 'summary': data.get('wiki').get('summary').split("