from aiohttp import web from aiohttp_apispec import docs, json_schema, querystring_schema from server.spotify import deserializers from server.spotify.constants import AlbumField, PlaylistField @docs( tags=["spotify"], summary="Get Spotify catalog information for a single track identified by its unique Spotify ID." ) @querystring_schema(deserializers.TrackDeserializer) async def track_details_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] track_id = request.match_info.get("track_id") data = request["querystring"] result = await api.track(track_id, data.get("market")) return web.json_response(result) @docs( tags=["spotify"], summary="Get Spotify catalog information for multiple tracks based on their Spotify IDs or ISRCs." ) @querystring_schema(deserializers.TracksDeserializer) async def tracks_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["querystring"] isrc_list = data.get("isrc_list") result = await api.tracks(data.get("ids") or isrc_list, data.get("market"), by_isrc=bool(isrc_list)) return web.json_response(result) @docs( tags=["spotify"], summary="Get multiple track's images by using Spotify catalog information based on their Spotify IDs.", ) @json_schema(deserializers.TracksImagesDeserializer) async def tracks_images_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["json"] result = await api.tracks_images(data["ids"], data["image_size"], data.get("market")) return web.json_response(result) @docs(tags=["spotify"], summary="Get a playlist owned by a Spotify user.") @querystring_schema(deserializers.PlaylistDeserializer) async def playlist_details_view(request: web.Request) -> web.Response: playlist_id = request.match_info.get("playlist_id") api = request.app["spotify_api"] data = request["querystring"] items_fields = data.get("items_fields") market = data.get("market") playlist = await api.playlist(playlist_id, market, data.get("fields"), data.get("additional_types")) if items_fields and PlaylistField.TRACKLIST_FULL in items_fields: playlist_tracks = await api.playlist_tracks( playlist_id, market, None, None, data.get("offset"), data.get("limit") ) if isinstance(playlist_tracks, dict) and "items" in playlist_tracks: playlist_tracks = playlist_tracks["items"] playlist["tracks"] = {"items": playlist_tracks, "total": playlist.get("tracks", {}).get("total", 0)} return web.json_response(playlist) @docs(tags=["spotify"], summary="Get playlist tracks.") @querystring_schema(deserializers.PlaylistTracksDeserializer) async def playlist_tracks_view(request: web.Request) -> web.Response: playlist_id = request.match_info.get("playlist_id") api = request.app["spotify_api"] data = request["querystring"] result = await api.playlist_tracks( playlist_id, data.get("market"), data.get("fields"), data.get("additional_types"), data.get("offset", 0), data.get("limit"), ) return web.json_response(result) @docs(tags=["spotify"], summary="Get multiple playlist owned by a Spotify user.") @querystring_schema(deserializers.PlaylistsDeserializer) async def playlists_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["querystring"] result = await api.playlists(data["ids"], data.get("market"), data.get("fields"), data.get("additional_types")) return web.json_response(result) @docs( tags=["spotify"], summary="Get multiple playlist's images by using Spotify catalog information based on their Spotify IDs.", ) @json_schema(deserializers.PlaylistsImagesDeserializer) async def playlists_images_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["json"] result = await api.playlists_images(data["ids"], data.get("market")) return web.json_response(result) @docs( tags=["spotify"], summary="Get information about albums, artists, playlists, tracks, shows or episodes that match a keyword string.", ) @querystring_schema(deserializers.SearchDeserializer) async def search_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["querystring"] result = await api.search( data.get("q") or data.get("query"), data.get("market"), data.get("type") or data.get("item_type"), data.get("include_external"), offset=data["offset"], limit=data["limit"], ) return web.json_response(result) @docs(tags=["spotify"], summary="Get Spotify catalog information for a single album with its tracks if needed.") @querystring_schema(deserializers.AlbumDeserializer) async def album_details_view(request: web.Request) -> web.Response: album_id = request.match_info.get("album_id") api = request.app["spotify_api"] data = request["querystring"] market = data.get("market") fields = data.get("fields") album = await api.album(album_id, market) if fields and (AlbumField.TRACKS in fields or AlbumField.TRACKS_FULL in fields): album_tracks = await api.album_tracks(album_id, market, data.get("offset"), data.get("limit")) if isinstance(album_tracks, dict) and "items" in album_tracks: album_tracks = album_tracks["items"] if AlbumField.TRACKS_FULL in fields: track_ids = [t["id"] for t in album_tracks] album_tracks = (await api.tracks(track_ids, market))["tracks"] album["tracks"] = album_tracks return web.json_response(album) @docs(tags=["spotify"], summary="Get Spotify catalog information for multiple albums identified by their Spotify IDs.") @querystring_schema(deserializers.AlbumsDeserializer) async def albums_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["querystring"] result = await api.albums(data["ids"], data.get("market")) return web.json_response(result) @docs( tags=["spotify"], summary="Get Spotify catalog information for a single artist identified by their unique Spotify ID.", ) async def artist_details_view(request: web.Request) -> web.Response: artist_id = request.match_info.get("artist_id") api = request.app["spotify_api"] result = await api.artist(artist_id) return web.json_response(result) @docs(tags=["spotify"], summary="Get Spotify catalog information for several artists based on their Spotify IDs.") @querystring_schema(deserializers.ArtistsDeserializer) async def artists_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] data = request["querystring"] result = await api.artists(data["ids"]) return web.json_response(result) @docs(tags=["spotify"], summary="Get public profile information about a Spotify user.") async def user_details_view(request: web.Request) -> web.Response: user_id = request.match_info.get("user_id") api = request.app["spotify_api"] result = await api.user(user_id) return web.json_response(result) @docs( tags=["spotify"], summary="Get supported markets.", ) async def markets_view(request: web.Request) -> web.Response: api = request.app["spotify_api"] result = await api.markets() return web.json_response({"markets": result})