spotify code cleanup

This commit is contained in:
vveeps 2021-10-28 16:48:25 +03:00
parent 5bc27bad58
commit df9ae54e6c
6 changed files with 85 additions and 86 deletions

View File

@ -1,6 +1,6 @@
"""Spotify module for Pomice, made possible by cloudwithax 2021""" """Spotify module for Pomice, made possible by cloudwithax 2021"""
from .exceptions import SpotifyRequestException from .exceptions import InvalidSpotifyURL, SpotifyRequestException
from .track import Track from .track import Track
from .playlist import Playlist from .playlist import Playlist
from .album import Album from .album import Album

View File

@ -1,15 +1,20 @@
from .track import Track from .track import Track
class Album: class Album:
"""The base class for a Spotify album""" """The base class for a Spotify album"""
def __init__(self, data: dict) -> None: def __init__(self, data: dict) -> None:
self.name = data['name'] self.name = data["name"]
self.artists = ", ".join(artist["name"] for artist in data['artists']) self.artists = ", ".join(artist["name"] for artist in data["artists"])
self.tracks = [Track(track) for track in data['tracks']['items']] self.tracks = [Track(track) for track in data["tracks"]["items"]]
self.total_tracks = data['total_tracks'] self.total_tracks = data["total_tracks"]
self.id = data['id'] self.id = data["id"]
self.image = data['images'][0]['url'] self.image = data["images"][0]["url"]
self.uri = data['external_urls']['spotify'] self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str: def __repr__(self) -> str:
return f"<Pomice.spotify.Album name={self.name} artists={self.artists} id={self.id} total_tracks={self.total_tracks} tracks={self.tracks}>" return (
f"<Pomice.spotify.Album name={self.name} artists={self.artists} id={self.id} "
f"total_tracks={self.total_tracks} tracks={self.tracks}>"
)

View File

@ -1,50 +1,55 @@
import base64
import re import re
import time import time
from base64 import b64encode
import aiohttp import aiohttp
from .album import Album from .album import Album
from .exceptions import SpotifyRequestException from .exceptions import InvalidSpotifyURL, SpotifyRequestException
from .playlist import Playlist from .playlist import Playlist
from .track import Track from .track import Track
GRANT_URL = "https://accounts.spotify.com/api/token" GRANT_URL = "https://accounts.spotify.com/api/token"
REQUEST_URL = "https://api.spotify.com/v1/{type}s/{id}"
SPOTIFY_URL_REGEX = re.compile( SPOTIFY_URL_REGEX = re.compile(
r"https?://open.spotify.com/(?P<type>album|playlist|track)/(?P<id>[a-zA-Z0-9]+)" r"https?://open.spotify.com/(?P<type>album|playlist|track)/(?P<id>[a-zA-Z0-9]+)"
) )
class Client: class Client:
"""The base client for the Spotify module of Pomice. """The base client for the Spotify module of Pomice.
This class will do all the heavy lifting of getting all the metadata This class will do all the heavy lifting of getting all the metadata
for any Spotify URL you throw at it. for any Spotify URL you throw at it.
""" """
def __init__(self, client_id: str, client_secret: str) -> None: def __init__(self, client_id: str, client_secret: str) -> None:
self._client_id: str = client_id self._client_id = client_id
self._client_secret: str = client_secret self._client_secret = client_secret
self.session = aiohttp.ClientSession() self.session = aiohttp.ClientSession()
self._bearer_token: str = None self._bearer_token: str = None
self._expiry: int = 0 self._expiry = 0
self._auth_token = base64.b64encode(":".join((self._client_id, self._client_secret)).encode()) self._auth_token = b64encode(f"{self._client_id}:{self._client_secret}".encode())
self._grant_headers = {"Authorization": f"Basic {self._auth_token.decode()}"} self._grant_headers = {"Authorization": f"Basic {self._auth_token.decode()}"}
self._bearer_headers = None self._bearer_headers = None
async def _fetch_bearer_token(self) -> None: async def _fetch_bearer_token(self) -> None:
data = {"grant_type": "client_credentials"} _data = {"grant_type": "client_credentials"}
async with self.session.post(GRANT_URL, data=data, headers=self._grant_headers) as resp:
if resp.status != 200: async with self.session.post(GRANT_URL, data=_data, headers=self._grant_headers) as resp:
raise SpotifyRequestException(f"Error: {resp.status} {resp.reason}") if resp.status != 200:
raise SpotifyRequestException(
f"Error fetching bearer token: {resp.status} {resp.reason}"
)
data: dict = await resp.json()
data = await resp.json()
self._bearer_token = data["access_token"] self._bearer_token = data["access_token"]
self._expiry = time.time() + (int(data["expires_in"]) - 10) self._expiry = time.time() + (int(data["expires_in"]) - 10)
self._bearer_headers = {"Authorization": f"Bearer {self._bearer_token}"} self._bearer_headers = {"Authorization": f"Bearer {self._bearer_token}"}
async def search(self, *, query: str): async def search(self, *, query: str):
if not self._bearer_token or time.time() >= self._expiry: if not self._bearer_token or time.time() >= self._expiry:
await self._fetch_bearer_token() await self._fetch_bearer_token()
@ -53,60 +58,37 @@ class Client:
spotify_id = result.group("id") spotify_id = result.group("id")
if not result: if not result:
return SpotifyRequestException("The Spotify link provided is not valid.") return InvalidSpotifyURL("The Spotify link provided is not valid.")
request_url = REQUEST_URL.format(type=spotify_type, id=spotify_id)
if spotify_type == "track":
request_url = f"https://api.spotify.com/v1/tracks/{spotify_id}"
async with self.session.get(request_url, headers=self._bearer_headers) as resp: async with self.session.get(request_url, headers=self._bearer_headers) as resp:
if resp.status != 200: if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason) raise SpotifyRequestException(
f"Error while fetching results: {resp.status} {resp.reason}"
)
data: dict = await resp.json() data: dict = await resp.json()
if spotify_type == "track":
return Track(data) return Track(data)
elif spotify_type == "album": elif spotify_type == "album":
request_url = f"https://api.spotify.com/v1/albums/{spotify_id}" return Album(data)
async with self.session.get(request_url, headers=self._bearer_headers) as resp: # processing a playlist result
if resp.status != 200: tracks = [Track(track["track"]) for track in data["tracks"]["items"]]
raise SpotifyRequestException(resp.status, resp.reason) next_page_url = data["tracks"]["next"]
album_data: dict = await resp.json() while next_page_url is not None:
return Album(album_data)
elif spotify_type == "playlist":
request_url = f"https://api.spotify.com/v1/playlists/{spotify_id}"
tracks = []
async with self.session.get(request_url, headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason)
playlist_data: dict = await resp.json()
tracks += [Track(track["track"]) for track in playlist_data["tracks"]["items"]]
next_page_url = playlist_data["tracks"]["next"]
while next_page_url != None:
async with self.session.get(next_page_url, headers=self._bearer_headers) as resp: async with self.session.get(next_page_url, headers=self._bearer_headers) as resp:
if resp.status != 200: if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason) raise SpotifyRequestException(
f"Error while fetching results: {resp.status} {resp.reason}"
next_page_data: dict = await resp.json() )
tracks += [Track(track["track"]) for track in next_page_data["items"]]
next_page_url = next_page_data["next"]
return Playlist(playlist_data, tracks)
next_data: dict = await resp.json()
tracks += [Track(track["track"]) for track in next_data["items"]]
next_page_url = next_data["next"]
return Playlist(data, tracks)

View File

@ -1,3 +1,7 @@
class SpotifyRequestException(Exception): class SpotifyRequestException(Exception):
"""An error occurred when making a request to the Spotify API""" """An error occurred when making a request to the Spotify API"""
pass pass
class InvalidSpotifyURL(Exception):
pass

View File

@ -1,16 +1,21 @@
from .track import Track from .track import Track
from typing import List from typing import List
class Playlist: class Playlist:
"""The base class for a Spotify playlist""" """The base class for a Spotify playlist"""
def __init__(self, data: dict, tracks: List[Track]) -> None: def __init__(self, data: dict, tracks: List[Track]) -> None:
self.name = data['name'] self.name = data["name"]
self.tracks = tracks self.tracks = tracks
self.owner = data['owner']['display_name'] self.owner = data["owner"]["display_name"]
self.total_tracks = data['tracks']['total'] self.total_tracks = data["tracks"]["total"]
self.id = data['id'] self.id = data["id"]
self.image = data['images'][0]['url'] self.image = data["images"][0]["url"]
self.uri = data['external_urls']['spotify'] self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str: def __repr__(self) -> str:
return f"<Pomice.spotify.Playlist name={self.name} owner={self.owner} id={self.id} total_tracks={self.total_tracks} tracks={self.tracks}>" return (
f"<Pomice.spotify.Playlist name={self.name} owner={self.owner} id={self.id} "
f"total_tracks={self.total_tracks} tracks={self.tracks}>"
)

View File

@ -2,17 +2,20 @@ class Track:
"""The base class for a Spotify Track""" """The base class for a Spotify Track"""
def __init__(self, data: dict) -> None: def __init__(self, data: dict) -> None:
self.name = data['name'] self.name = data["name"]
self.artists = ", ".join(artist["name"] for artist in data['artists']) self.artists = ", ".join(artist["name"] for artist in data["artists"])
self.length = data['duration_ms'] self.length = data["duration_ms"]
self.id = data['id'] self.id = data["id"]
if data.get("album") and data["album"]["images"]: if data.get("album") and data["album"]["images"]:
self.image = data['album']['images'][0]['url'] self.image = data["album"]["images"][0]["url"]
else: else:
self.image = None self.image = None
self.uri = data['external_urls']['spotify'] self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str: def __repr__(self) -> str:
return f"<Pomice.spotify.Track name={self.name} artists={self.artists} length={self.length} id={self.id}>" return (
f"<Pomice.spotify.Track name={self.name} artists={self.artists} "
f"length={self.length} id={self.id}>"
)