api/endpoints/lastfm.py

#!/usr/bin/env python3.12

import importlib
import traceback
from typing import Optional, Union
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from .constructors import ValidArtistSearchRequest, ValidAlbumDetailRequest,\
    ValidTrackInfoRequest, LastFMException

class LastFM(FastAPI):
    """Last.FM Endpoints"""   
    def __init__(self, app: FastAPI,
                 util, constants) -> None:
        self.app: FastAPI = app
        self.util = util
        self.constants = constants
        self.lastfm = importlib.import_module("lastfm_wrapper").LastFM()

        self.endpoints: dict = {
            "lastfm/get_artist_by_name": self.artist_by_name_handler,
            "lastfm/get_artist_albums": self.artist_album_handler,
            "lastfm/get_release": self.release_detail_handler,
            "lastfm/get_release_tracklist": self.release_tracklist_handler,
            "lastfm/get_track_info": self.track_info_handler,
            #tbd
            }

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(f"/{endpoint}", handler, methods=["POST"],
                              include_in_schema=True)
    
    async def artist_by_name_handler(self, data: ValidArtistSearchRequest) -> JSONResponse:
        """
        Get artist info
        - **a**: Artist to search
        """
        artist: Optional[str] = data.a.strip()
        if not artist:
            return JSONResponse(content={
                'err': True,
                'errorText': 'No artist specified',
                })
        
        artist_result = await self.lastfm.search_artist(artist=artist)
        if not artist_result or "err" in artist_result.keys():
            return JSONResponse(status_code=500, content={
                'err': True,
                'errorText': 'Search failed (no results?)',
                })
        
        return JSONResponse(content={
            'success': True,
            'result': artist_result,
            })
    
    async def artist_album_handler(self, data: ValidArtistSearchRequest) -> JSONResponse:
        """
        Get artist's albums/releases
        - **a**: Artist to search
        """
        artist: str = data.a.strip()
        if not artist:
            return JSONResponse(status_code=500, content={
                'err': True,
                'errorText': 'Invalid request: No artist specified',
                })
        
        album_result: Union[dict, list[dict]] = await self.lastfm.get_artist_albums(artist=artist)
        if isinstance(album_result, dict):
            return JSONResponse(status_code=500, content={
                'err': True,
                'errorText': 'General failure.',
            })
        album_result_out: list = []
        seen_release_titles: list = []
        
        for release in album_result:
            release_title: str = release.get('title', 'Unknown')
            if release_title.lower() in seen_release_titles:
                continue
            seen_release_titles.append(release_title.lower())
            album_result_out.append(release)

        return JSONResponse(content={
            'success': True,
            'result': album_result_out
            })

    async def release_detail_handler(self, data: ValidAlbumDetailRequest) -> JSONResponse:
        """
        Get details of a particular release by an artist
        - **a**: Artist to search
        - **release**: Release title to search
        """
        artist: str = data.a.strip()
        release: str = data.release.strip()
        
        if not artist or not release:
            return JSONResponse(status_code=500, content={
                'err': True,
                'errorText': 'Invalid request',
                })
        
        release_result = await self.lastfm.get_release(artist=artist, album=release)
        ret_obj = {
            'id': release_result.get('id'),
            'artists': release_result.get('artists'),
            'title': release_result.get('title'),
            'summary': release_result.get('summary'),
            'tracks': release_result.get('tracks'),
            } 

        return JSONResponse(content={
            'success': True,
            'result': ret_obj,
            })
    
    async def release_tracklist_handler(self, data: ValidAlbumDetailRequest) -> JSONResponse:
        """
        Get track list for a particular release by an artist
        - **a**: Artist to search
        - **release**: Release title to search 
        """
        artist: str = data.a.strip()
        release: str = data.release.strip()
        
        if not artist or not release:
            return JSONResponse(status_code=500, content={
                'err': True,
                'errorText': 'Invalid request',
            })
        
        tracklist_result: dict  = await self.lastfm.get_album_tracklist(artist=artist, album=release)
        return JSONResponse(content={
            'success': True,
            'id': tracklist_result.get('id'),
            'artists': tracklist_result.get('artists'),
            'title': tracklist_result.get('title'),
            'summary': tracklist_result.get('summary'),
            'tracks': tracklist_result.get('tracks'),
            })

    async def track_info_handler(self, data: ValidTrackInfoRequest) -> JSONResponse:
        """
         Get track info from Last.FM given an artist/track
         - **a**: Artist to search
         - **t**: Track title to search
         """    
        try:
            artist: str = data.a
            track: str = data.t
            
            if not artist or not track:
                return JSONResponse(status_code=500, content={
                    'err': True,
                    'errorText': 'Invalid request'
                    })
            
            track_info_result: dict = await self.lastfm.get_track_info(artist=artist,
                                                                       track=track)
            if "err" in track_info_result:
                raise LastFMException("Unknown error occurred: %s", 
                                      track_info_result.get('errorText', '??'))
            return JSONResponse(content={
                'success': True,
                'result': track_info_result
            })
        except:
            traceback.print_exc()
            return JSONResponse(status_code=500, content={
                'err': True,
                'errorText': 'General error',
            })