nambrogi
/
pomice
mirror of https://github.com/cloudwithax/pomice.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Major Spotify impl rewrite + some other goodies

This commit is contained in:
cloudwithax 2021-10-19 20:29:24 -04:00
parent ea6c2baf3c
commit 7d53934697
31 changed files with 308 additions and 4997 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -8,7 +8,6 @@ The modern [Lavalink](https://github.com/freyacodes/Lavalink) wrapper designed f
This library is heavily based off of/uses code from the following libraries:
- [Wavelink](https://github.com/PythonistaGuild/Wavelink)
- [spotify.py](https://github.com/mental32/spotify.py)
- [Slate](https://github.com/Axelancerr/slate)
- [Granitepy](https://github.com/twitch0001/granitepy)

2
pomice/__init__.py
View File

@ -18,7 +18,9 @@ __author__ = "cloudwithax"
from .enums import SearchType
from .events import *
from .exceptions import *
from .spotify import *
from .filters import *
from .objects import *
from .player import Player
from .pool import *

67
pomice/events.py
View File

@ -1,3 +1,6 @@
import asyncio
from pomice import exceptions
from .pool import NodePool
@ -11,75 +14,75 @@ class PomiceEvent:
```
"""
name = "event"
class TrackStartEvent(PomiceEvent):
"""Fired when a track has successfully started.
Returns the player associated with the track and the track ID
Returns the player associated with the event and the pomice.Track object.
"""
def __init__(self, data):
def __init__(self, player, track):
super().__init__()
self.name = "track_start"
self.player = NodePool.get_node().get_player(int(data["guildId"]))
self.track_id = data["track"]
self.player = player
self.track = track
def __repr__(self) -> str:
return f"<Pomice.TrackStartEvent track_id={self.track_id}>"
return f"<Pomice.TrackStartEvent player={self.player} track_id={self.track.track_id}>"
class TrackEndEvent(PomiceEvent):
"""Fired when a track has successfully ended.
Returns the player associated with the track along with the track ID and reason.
Returns the player associated with the event along with the pomice.Track object and reason.
"""
def __init__(self, data):
def __init__(self, player, track, reason):
super().__init__()
self.name = "track_end"
self.player = NodePool.get_node().get_player(int(data["guildId"]))
self.track_id = data["track"]
self.reason = data["reason"]
self.player = player
self.track = track
self.reason = reason
def __repr__(self) -> str:
return f"<Pomice.TrackEndEvent track_id={self.track_id} reason={self.reason}>"
return f"<Pomice.TrackEndEvent player={self.player} track_id={self.track.track_id} reason={self.reason}>"
class TrackStuckEvent(PomiceEvent):
"""Fired when a track is stuck and cannot be played. Returns the player
associated with the track along with a track ID to be further parsed by the end user.
associated with the event along with the pomice.Track object to be further parsed by the end user.
"""
def __init__(self, data):
def __init__(self, player, track, threshold):
super().__init__()
self.name = "track_stuck"
self.player = NodePool.get_node().get_player(int(data["guildId"]))
self.player = player
self.track_id = data["track"]
self.threshold = data["thresholdMs"]
self.track = track
self.threshold = threshold
def __repr__(self) -> str:
return f"<Pomice.TrackStuckEvent track_id={self.track_id} threshold={self.threshold}>"
return f"<Pomice.TrackStuckEvent player={self.player} track_id={self.track.track_id} threshold={self.threshold}>"
class TrackExceptionEvent(PomiceEvent):
"""Fired when a track error has occured.
Returns the player associated with the track along with the error code and exception.
Returns the player associated with the event along with the error code and exception.
"""
def __init__(self, data):
def __init__(self, player, track, error):
super().__init__()
self.name = "track_exception"
self.player = NodePool.get_node().get_player(int(data["guildId"]))
self.error = data["error"]
self.exception = data["exception"]
self.player = player
self.track = track
self.error = error
def __repr__(self) -> str:
return f"<Pomice.TrackExceptionEvent> error={self.error} exeception={self.exception}"
return f"<Pomice.TrackExceptionEvent player={self.player} error={self.error} exeception={self.exception}>"
class WebSocketClosedEvent(PomiceEvent):
@ -87,16 +90,16 @@ class WebSocketClosedEvent(PomiceEvent):
Returns the reason and the error code.
"""
def __init__(self, data):
def __init__(self, guild, reason, code):
super().__init__()
self.name = "websocket_closed"
self.reason = data["reason"]
self.code = data["code"]
self.guild = guild
self.reason = reason
self.code = code
def __repr__(self) -> str:
return f"<Pomice.WebsocketClosedEvent reason={self.reason} code={self.code}>"
return f"<Pomice.WebsocketClosedEvent guild_id={self.guild.id} reason={self.reason} code={self.code}>"
class WebSocketOpenEvent(PomiceEvent):
@ -104,13 +107,13 @@ class WebSocketOpenEvent(PomiceEvent):
Returns the target and the session SSRC.
"""
def __init__(self, data):
def __init__(self, target, ssrc):
super().__init__()
self.name = "websocket_open"
self.target: str = data["target"]
self.ssrc: int = data["ssrc"]
self.target: str = target
self.ssrc: int = ssrc
def __repr__(self) -> str:
return f"<Pomice.WebsocketOpenEvent target={self.target} ssrc={self.ssrc}>"

1
pomice/exceptions.py
View File

@ -62,3 +62,4 @@ class SpotifyPlaylistLoadFailed(PomiceException):
class InvalidSpotifyClientAuthorization(PomiceException):
"""No Spotify client authorization was provided for track searching."""
pass

50
pomice/player.py
View File

@ -1,8 +1,18 @@
import time
from typing import Any, Dict, Optional, Type, Union
from typing import (
Any,
Dict,
Optional,
Type,
Union
)
import discord
from discord import Client, Guild, VoiceChannel, VoiceProtocol
from discord import (
Client,
Guild,
VoiceChannel,
VoiceProtocol
)
from discord.ext import commands
from pomice.enums import SearchType
@ -134,19 +144,47 @@ class Player(VoiceProtocol):
async def on_voice_state_update(self, data: dict):
self._voice_state.update({"sessionId": data.get("session_id")})
if not (channel_id := data.get("channel_id")):
self.channel = None
self._voice_state.clear()
return
self.channel = self.guild.get_channel(int(channel_id))
if not data.get('token'):
return
await self._dispatch_voice_update({**self._voice_state, "event": data})
async def _dispatch_event(self, data: dict):
event_type = data.get("type")
event = getattr(events, event_type, None)
event = event(data)
self.bot.dispatch(f"pomice_{event.name}", event)
if event_type == "TrackStartEvent":
track = await self._node.build_track(data["track"])
event = events.TrackStartEvent(self, track)
self.dispatch(event, self, track)
elif event_type == "TrackEndEvent":
track = await self._node.build_track(data["track"])
event = events.TrackEndEvent(self, track, data["reason"])
self.dispatch(event, self, track, data["reason"])
elif event_type == "TrackExceptionEvent":
track = await self._node.build_track(data["track"])
event = events.TrackExceptionEvent(self, track, data["error"])
self.dispatch(event, self, track, data["error"])
elif event_type == "TrackStuckEvent":
track = await self._node.build_track(data["track"])
event = events.TrackStuckEvent(self, track, data["thresholdMs"])
self.dispatch(event, self, track, data["thresholdMs"])
elif event_type == "WebSocketOpenEvent":
event = events.WebSocketOpenEvent(data["target"], data["ssrc"])
self.dispatch(event, data["target"], data["ssrc"])
elif event_type == "WebSocketClosedEvent":
event = events.WebSocketClosedEvent(self._guild, data["reason"], data["code"])
self.dispatch(event, self._guild, data["reason"], data["code"])
def dispatch(self, event, *args, **kwargs):
self.bot.dispatch(f"pomice_{event.name}", event, *args, **kwargs)
async def get_tracks(
self,

191
pomice/pool.py
View File

@ -8,7 +8,6 @@ import socket
import time
from typing import Dict, Optional, Type, TYPE_CHECKING
from urllib.parse import quote
from base64 import b
import aiohttp
import discord
@ -28,7 +27,6 @@ from .exceptions import (
TrackLoadError
)
from .objects import Playlist, Track
from .spotify import SpotifyException
from .utils import ExponentialBackoff, NodeStats
if TYPE_CHECKING:
@ -99,9 +97,6 @@ class Node:
self._spotify_client = spotify.Client(
self._spotify_client_id, self._spotify_client_secret
)
self._spotify_http_client = spotify.HTTPClient(
self._spotify_client_id, self._spotify_client_secret
)
self._bot.add_listener(self._update_handler, "on_socket_response")
@ -256,10 +251,9 @@ class Node:
async def build_track(
self,
*,
identifier: str,
ctx: Optional[commands.Context]
):
ctx: Optional[commands.Context] = None
) -> Track:
"""
Builds a track using a valid track identifier
@ -267,7 +261,7 @@ class Node:
Context object on the track it builds.
"""
async with self.session.get(f'{self._rest_uri}/decodetrack?',
async with self._session.get(f'{self._rest_uri}/decodetrack?',
headers={'Authorization': self._password},
params={'track': identifier}) as resp:
@ -277,8 +271,7 @@ class Node:
raise TrackLoadError(f'Failed to build track. Status: {data["status"]}, Error: {data["error"]}.'
f'Check the identifier is correct and try again.')
track = Track(track_id=identifier, ctx=ctx, info=data)
return track
return Track(track_id=identifier, ctx=ctx, info=data)
async def get_tracks(
@ -300,7 +293,7 @@ class Node:
if not URL_REGEX.match(query) and not re.match(r"(?:ytm?|sc)search:.", query):
query = f"{search_type}:{query}"
if spotify_url_check := SPOTIFY_URL_REGEX.match(query):
if SPOTIFY_URL_REGEX.match(query):
if not self._spotify_client_id and not self._spotify_client_secret:
raise InvalidSpotifyClientAuthorization(
"You did not provide proper Spotify client authorization credentials. "
@ -308,120 +301,92 @@ class Node:
"please obtain Spotify API credentials here: https://developer.spotify.com/"
)
spotify_search_type = spotify_url_check.group("type")
spotify_id = spotify_url_check.group("id")
spotify_results = await self._spotify_client.search(query=query)
if spotify_search_type == "playlist":
results = spotify.Playlist(
client=self._spotify_client,
data=await self._spotify_http_client.get_playlist(spotify_id)
if isinstance(spotify_results, spotify.Playlist):
tracks = [
Track(
track_id=track.id,
ctx=ctx,
search_type=search_type,
spotify=True,
info={
"title": track.name,
"author": track.artists,
"length": track.length,
"identifier": track.id,
"uri": track.uri,
"isStream": False,
"isSeekable": False,
"position": 0,
"thumbnail": track.image
},
) for track in spotify_results.tracks
]
return Playlist(
playlist_info={"name": spotify_results.name, "selectedTrack": tracks[0]},
tracks=tracks,
ctx=ctx,
spotify=True,
thumbnail=spotify_results.image,
uri=spotify_results.uri,
)
try:
search_tracks = await results.get_all_tracks()
tracks = [
Track(
track_id=track.id,
ctx=ctx,
search_type=search_type,
spotify=True,
info={
"title": track.name or "Unknown",
"author": ", ".join(
artist.name for artist in track.artists
) or "Unknown",
"length": track.duration or 0,
"identifier": track.id or "Unknown",
"uri": track.url or "spotify",
"isStream": False,
"isSeekable": False,
"position": 0,
"thumbnail": track.images[0].url if track.images else None
},
) for track in search_tracks
]
return Playlist(
playlist_info={"name": results.name, "selectedTrack": tracks[0]},
tracks=tracks,
elif isinstance(spotify_results, spotify.Album):
tracks = [
Track(
track_id=track.id,
ctx=ctx,
search_type=search_type,
spotify=True,
thumbnail=results.images[0].url,
uri=results.url,
)
info={
"title": track.name,
"author": track.artists,
"length": track.length,
"identifier": track.id,
"uri": track.uri,
"isStream": False,
"isSeekable": False,
"position": 0,
"thumbnail": track.image
},
) for track in spotify_results.tracks
]
except SpotifyException:
raise SpotifyPlaylistLoadFailed(
f"Unable to find results for {query}"
)
return Playlist(
playlist_info={"name": spotify_results.name, "selectedTrack": tracks[0]},
tracks=tracks,
ctx=ctx,
spotify=True,
thumbnail=spotify_results.image,
uri=spotify_results.uri,
)
elif spotify_search_type == "album":
results = await self._spotify_client.get_album(spotify_id=spotify_id)
try:
search_tracks = await results.get_all_tracks()
tracks = [
Track(
track_id=track.id,
ctx=ctx,
search_type=search_type,
spotify=True,
info={
"title": track.name or "Unknown",
"author": ", ".join(
artist.name for artist in track.artists
) or "Unknown",
"length": track.duration or 0,
"identifier": track.id or "Unknown",
"uri": track.url or "spotify",
"isStream": False,
"isSeekable": False,
"position": 0,
"thumbnail": track.images[0].url if track.images else None
},
) for track in search_tracks
]
elif isinstance(spotify_results, spotify.Track):
return Playlist(
playlist_info={"name": results.name, "selectedTrack": tracks[0]},
tracks=tracks,
return [
Track(
track_id=spotify_results.id,
ctx=ctx,
search_type=search_type,
spotify=True,
thumbnail=results.images[0].url,
uri=results.url,
info={
"title": spotify_results.name,
"author": spotify_results.artists,
"length": spotify_results.length,
"identifier": spotify_results.id,
"uri": spotify_results.uri,
"isStream": False,
"isSeekable": False,
"position": 0,
"thumbnail": spotify_results.image
},
)
]
except SpotifyException:
raise SpotifyAlbumLoadFailed(f"Unable to find results for {query}")
elif spotify_search_type == 'track':
try:
results = await self._spotify_client.get_track(spotify_id=spotify_id)
return [
Track(
track_id=results.id,
ctx=ctx,
search_type=search_type,
spotify=True,
info={
"title": results.name or "Unknown",
"author": ", ".join(
artist.name for artist in results.artists
) or "Unknown",
"length": results.duration or 0,
"identifier": results.id or "Unknown",
"uri": results.url or "spotify",
"isStream": False,
"isSeekable": False,
"position": 0,
"thumbnail": results.images[0].url if results.images else None
},
)
]
except SpotifyException:
raise SpotifyTrackLoadFailed(f"Unable to find results for {query}")
elif discord_url := DISCORD_MP3_URL_REGEX.match(query):
async with self._session.get(

33
pomice/spotify/__init__.py
View File

@ -1,28 +1,7 @@
__version__ = "0.10.2"
__title__ = "spotify"
__author__ = "mental"
__license__ = "MIT"
"""Spotify module for Pomice, made possible by cloudwithax 2021"""
from typing import Dict, Type
from .oauth import *
from .utils import clean as _clean_namespace
from .errors import *
from .models import *
from .client import *
from .models import SpotifyBase
from .http import HTTPClient, HTTPUserClient
__all__ = tuple(name for name in locals() if name[0] != "_")
_locals = locals() # pylint: disable=invalid-name
_types: Dict[str, Type[Union[SpotifyBase, HTTPClient]]]
with _clean_namespace(locals(), "_locals", "_clean_namespace"):
_types = dict( # pylint: disable=invalid-name
(name, _locals[name])
for name, obj in _locals.items()
if isinstance(obj, type) and issubclass(obj, SpotifyBase)
)
_types["HTTPClient"] = HTTPClient
_types["HTTPUserClient"] = HTTPUserClient
from .exceptions import SpotifyRequestException
from .track import Track
from .playlist import Playlist
from .album import Album
from .client import Client

15
pomice/spotify/album.py Normal file
View File

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

408
pomice/spotify/client.py
View File

@ -1,348 +1,130 @@
import asyncio
from typing import Optional, List, Iterable, NamedTuple, Type, Union, Dict
import aiohttp
import re
import time
import base64
from .http import HTTPClient
from .utils import to_id
from . import OAuth2, Artist, Album, Track, User, Playlist
__all__ = ("Client", "SearchResults")
from .exceptions import SpotifyRequestException
from .album import Album
from .playlist import Playlist
from .track import Track
_TYPES = {"artist": Artist, "album": Album, "playlist": Playlist, "track": Track}
_SEARCH_TYPES = {"track", "playlist", "artist", "album"}
_SEARCH_TYPE_ERR = (
'Bad queary type! got "%s" expected any of: track, playlist, artist, album'
GRANT_URL = 'https://accounts.spotify.com/api/token'
SPOTIFY_URL_REGEX = re.compile(
r"https?://open.spotify.com/(?P<type>album|playlist|track)/(?P<id>[a-zA-Z0-9]+)"
)
class SearchResults(NamedTuple):
"""A namedtuple of search results.
Attributes
----------
artists : List[:class:`Artist`]
The artists of the search.
playlists : List[:class:`Playlist`]
The playlists of the search.
albums : List[:class:`Album`]
The albums of the search.
tracks : List[:class:`Track`]
The tracks of the search.
"""
artists: Optional[List[Artist]] = None
playlists: Optional[List[Playlist]] = None
albums: Optional[List[Album]] = None
tracks: Optional[List[Track]] = None
class Client:
"""Represents a Client app on Spotify.
This class is used to interact with the Spotify API.
Parameters
----------
client_id : :class:`str`
The client id provided by spotify for the app.
client_secret : :class:`str`
The client secret for the app.
loop : Optional[:class:`asyncio.AbstractEventLoop`]
The event loop the client should run on, if no loop is specified `asyncio.get_event_loop()` is called and used instead.
Attributes
----------
client_id : :class:`str`
The applications client_id, also aliased as `id`
http : :class:`HTTPClient`
The HTTPClient that is being used.
loop : Optional[:class:`asyncio.AbstractEventLoop`]
The event loop the client is running on.
"""
The base client for the Spotify module of Pomice.
This class will do all the heavy lifting of getting all the metadata for any Spotify URL you throw at it.
"""
def __init__(self, client_id: str, client_secret: str) -> None:
print("Client initialized")
self._client_id: str = client_id
self._client_secret: str = client_secret
_default_http_client: Type[HTTPClient] = HTTPClient
self.session = aiohttp.ClientSession()
def __init__(
self,
client_id: str,
client_secret: str,
*,
loop: Optional[asyncio.AbstractEventLoop] = None,
) -> None:
if not isinstance(client_id, str):
raise TypeError("client_id must be a string.")
self._bearer_token: str = None
self._expiry: int = 0
self._auth_token = base64.b64encode(":".join((self._client_id, self._client_secret)).encode())
self._grant_headers = {"Authorization": f"Basic {self._auth_token.decode()}"}
self._bearer_headers = None
if not isinstance(client_secret, str):
raise TypeError("client_secret must be a string.")
if loop is not None and not isinstance(loop, asyncio.AbstractEventLoop):
raise TypeError(
"loop argument must be None or an instance of asyncio.AbstractEventLoop."
)
async def _fetch_bearer_token(self) -> None:
data = {"grant_type": "client_credentials"}
async with self.session.post(GRANT_URL, data=data, headers=self._grant_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(f"Error: {resp.status} {resp.reason}")
self.loop = loop = loop or asyncio.get_event_loop()
self.http = self._default_http_client(client_id, client_secret, loop=loop)
data = await resp.json()
self._bearer_token = data['access_token']
self._expiry = time.time() + (int(data['expires_in']) - 10)
self._bearer_headers = {'Authorization': f'Bearer {self._bearer_token}'}
def __repr__(self):
return f"<spotify.Client: {self.http.client_id!r}>"
async def __aenter__(self) -> "Client":
return self
async def search(self, *, query: str):
if not self._bearer_token or time.time() >= self._expiry:
await self._fetch_bearer_token()
async def __aexit__(self, exc_type, exc_value, traceback) -> None:
await self.close()
result = SPOTIFY_URL_REGEX.match(query)
spotify_type = result.group('type')
spotify_id = result.group('id')
# Properties
if not result:
return SpotifyRequestException("The Spotify link provided is not valid.")
@property
def client_id(self) -> str:
""":class:`str` - The Spotify client ID."""
return self.http.client_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:
if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason)
@property
def id(self): # pylint: disable=invalid-name
""":class:`str` - The Spotify client ID."""
return self.http.client_id
data: dict = await resp.json()
# Public api
return Track(data)
elif spotify_type == "album":
request_url = f"https://api.spotify.com/v1/albums/{spotify_id}"
def oauth2_url(
self,
redirect_uri: str,
scopes: Optional[Union[Iterable[str], Dict[str, bool]]] = None,
state: Optional[str] = None,
) -> str:
"""Generate an oauth2 url for user authentication.
async with self.session.get(request_url, headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason)
This is an alias to :meth:`OAuth2.url_only` but the
difference is that the client id is autmatically
passed in to the constructor.
album_data: dict = await resp.json()
Parameters
----------
redirect_uri : :class:`str`
Where spotify should redirect the user to after authentication.
scopes : Optional[Iterable[:class:`str`], Dict[:class:`str`, :class:`bool`]]
The scopes to be requested.
state : Optional[:class:`str`]
Using a state value can increase your assurance that an incoming connection is the result of an
authentication request.
return Album(album_data)
Returns
-------
url : :class:`str`
The OAuth2 url.
"""
return OAuth2.url_only(
client_id=self.http.client_id,
redirect_uri=redirect_uri,
scopes=scopes,
state=state,
)
elif spotify_type == "playlist":
# Okay i know this looks like a mess, but hear me out, this works
# The Spotify Web API limits how many tracks can be seen in a single request to 100
# So we have to do some clever techniques to get all the tracks in any playlist larger than 100 songs
# This method doesn't need to be applied to albums due to the fact that 99% of albums
# are never more than 100 tracks (I'm looking at you, Deep Zone Project...)
request_url = f"https://api.spotify.com/v1/playlists/{spotify_id}"
# Set the offset now so we can change it when we get all the tracks
offset = 0
tracks = []
async def close(self) -> None:
"""Close the underlying HTTP session to Spotify."""
await self.http.close()
# First, get the playlist data so we can get the total amount of tracks for later
async with self.session.get(request_url, headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason)
async def user_from_token(self, token: str) -> User:
"""Create a user session from a token.
playlist_data: dict = await resp.json()
.. note::
# Second, get the total amount of tracks in said playlist so we can use this to get all the tracks
total_tracks: int = playlist_data['tracks']['total']
This code is equivelent to `User.from_token(client, token)`
# This section of code may look spammy, but trust me, it's not
while len(tracks) < total_tracks:
tracks_request_url = f"https://api.spotify.com/v1/playlists/{spotify_id}/tracks?offset={offset}&limit=100"
async with self.session.get(tracks_request_url, headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(resp.status, resp.reason)
Parameters
----------
token : :class:`str`
The token to attatch the user session to.
playlist_track_data: dict = await resp.json()
Returns
-------
user : :class:`spotify.User`
The user from the ID
"""
return await User.from_token(self, token)
# This is the juicy part..
# Add the tracks we got from the current page of results
tracks += [Track(track['track']) for track in playlist_track_data['items']]
# Set the offset to go to the next page
offset += 100
# Repeat until we have all the tracks
# We have all the tracks, cast to the class for easier reading
return Playlist(playlist_data, tracks)
async def get_album(self, spotify_id: str, *, market: str = "US") -> Album:
"""Retrive an album with a spotify ID.
Parameters
----------
spotify_id : :class:`str`
The ID to search for.
market : Optional[:class:`str`]
An ISO 3166-1 alpha-2 country code
Returns
-------
album : :class:`spotify.Album`
The album from the ID
"""
data = await self.http.album(to_id(spotify_id), market=market)
return Album(self, data)
async def get_artist(self, spotify_id: str) -> Artist:
"""Retrive an artist with a spotify ID.
Parameters
----------
spotify_id : str
The ID to search for.
Returns
-------
artist : Artist
The artist from the ID
"""
data = await self.http.artist(to_id(spotify_id))
return Artist(self, data)
async def get_track(self, spotify_id: str) -> Track:
"""Retrive an track with a spotify ID.
Parameters
----------
spotify_id : str
The ID to search for.
Returns
-------
track : Track
The track from the ID
"""
data = await self.http.track(to_id(spotify_id))
return Track(self, data)
async def get_user(self, spotify_id: str) -> User:
"""Retrive an user with a spotify ID.
Parameters
----------
spotify_id : str
The ID to search for.
Returns
-------
user : User
The user from the ID
"""
data = await self.http.user(to_id(spotify_id))
return User(self, data)
# Get multiple objects
async def get_albums(self, *ids: str, market: str = "US") -> List[Album]:
"""Retrive multiple albums with a list of spotify IDs.
Parameters
----------
ids : List[str]
the ID to look for
market : Optional[str]
An ISO 3166-1 alpha-2 country code
Returns
-------
albums : List[Album]
The albums from the IDs
"""
data = await self.http.albums(
",".join(to_id(_id) for _id in ids), market=market
)
return list(Album(self, album) for album in data["albums"])
async def get_artists(self, *ids: str) -> List[Artist]:
"""Retrive multiple artists with a list of spotify IDs.
Parameters
----------
ids : List[:class:`str`]
The IDs to look for.
Returns
-------
artists : List[:class:`Artist`]
The artists from the IDs
"""
data = await self.http.artists(",".join(to_id(_id) for _id in ids))
return list(Artist(self, artist) for artist in data["artists"])
async def search( # pylint: disable=invalid-name
self,
q: str,
*,
types: Iterable[str] = ("track", "playlist", "artist", "album"),
limit: int = 20,
offset: int = 0,
market: str = "US",
should_include_external: bool = False,
) -> SearchResults:
"""Access the spotify search functionality.
>>> results = client.search('Cadet', types=['artist'])
>>> for artist in result.get('artists', []):
... if artist.name.lower() == 'cadet':
... print(repr(artist))
... break
Parameters
----------
q : :class:`str`
the search query
types : Optional[Iterable[`:class:`str`]]
A sequence of search types (can be any of `track`, `playlist`, `artist` or `album`) to refine the search request.
A `ValueError` may be raised if a search type is found that is not valid.
limit : Optional[:class:`int`]
The limit of search results to return when searching.
Maximum limit is 50, any larger may raise a :class:`HTTPException`
offset : Optional[:class:`int`]
The offset from where the api should start from in the search results.
market : Optional[:class:`str`]
An ISO 3166-1 alpha-2 country code. Provide this parameter if you want to apply Track Relinking.
should_include_external : :class:`bool`
If `True` is specified, the response will include any relevant audio content
that is hosted externally. By default external content is filtered out from responses.
Returns
-------
results : :class:`SearchResults`
The results of the search.
Raises
------
TypeError
Raised when a parameter with a bad type is passed.
ValueError
Raised when a bad search type is passed with the `types` argument.
"""
if not hasattr(types, "__iter__"):
raise TypeError("types must be an iterable.")
types_ = set(types)
if not types_.issubset(_SEARCH_TYPES):
raise ValueError(_SEARCH_TYPE_ERR % types_.difference(_SEARCH_TYPES).pop())
query_type = ",".join(tp.strip() for tp in types)
include_external: Optional[str]
if should_include_external:
include_external = "audio"
else:
include_external = None
data = await self.http.search(
q=q,
query_type=query_type,
market=market,
limit=limit,
offset=offset,
include_external=include_external,
)
return SearchResults(
**{
key: [_TYPES[obj["type"]](self, obj) for obj in value["items"]]
for key, value in data.items()
}
)

41
pomice/spotify/errors.py
View File

@ -1,41 +0,0 @@
__all__ = ("SpotifyException", "HTTPException", "Forbidden", "NotFound")
class SpotifyException(Exception):
"""Base exception class for spotify.py."""
class HTTPException(SpotifyException):
"""A generic exception that's thrown when a HTTP operation fails."""
def __init__(self, response, message):
self.response = response
self.status = response.status
error = message.get("error")
if isinstance(error, dict):
self.text = error.get("message", "")
else:
self.text = message.get("error_description", "")
fmt = "{0.reason} (status code: {0.status})"
if self.text.strip():
fmt += ": {1}"
super().__init__(fmt.format(self.response, self.text))
class Forbidden(HTTPException):
"""An exception that's thrown when status code 403 occurs."""
class NotFound(HTTPException):
"""An exception that's thrown when status code 404 occurs."""
class BearerTokenError(HTTPException):
"""An exception that's thrown when Spotify could not provide a valid Bearer Token"""
class RateLimitedException(Exception):
"""An exception that gets thrown when a rate limit is encountered."""

3
pomice/spotify/exceptions.py Normal file
View File

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

1789
pomice/spotify/http.py
View File

File diff suppressed because it is too large Load Diff

26
pomice/spotify/models/__init__.py
View File

@ -1,26 +0,0 @@
from .. import _clean_namespace
from . import typing
from .base import AsyncIterable, SpotifyBase, URIBase
from .common import Device, Context, Image
from .artist import Artist
from .track import Track, PlaylistTrack
from .player import Player
from .album import Album
from .library import Library
from .playlist import Playlist
from .user import User
__all__ = (
"User",
"Track",
"PlaylistTrack",
"Artist",
"Album",
"Playlist",
"Library",
"Player",
"Device",
"Context",
"Image",
)

121
pomice/spotify/models/album.py
View File

@ -1,121 +0,0 @@
from functools import partial
from typing import Optional, List
from ..oauth import set_required_scopes
from . import AsyncIterable, URIBase, Image, Artist, Track
class Album(URIBase, AsyncIterable): # pylint: disable=too-many-instance-attributes
"""A Spotify Album.
Attributes
----------
artists : List[Artist]
The artists for the album.
id : str
The ID of the album.
name : str
The name of the album.
href : str
The HTTP API URL for the album.
uri : str
The URI for the album.
album_group : str
ossible values are “album”, “single”, “compilation”, “appears_on”.
Compare to album_type this field represents relationship between the artist and the album.
album_type : str
The type of the album: one of "album" , "single" , or "compilation".
release_date : str
The date the album was first released.
release_date_precision : str
The precision with which release_date value is known: year, month or day.
genres : List[str]
A list of the genres used to classify the album.
label : str
The label for the album.
popularity : int
The popularity of the album. The value will be between 0 and 100, with 100 being the most popular.
copyrights : List[Dict]
The copyright statements of the album.
markets : List[str]
The markets in which the album is available: ISO 3166-1 alpha-2 country codes.
"""
def __init__(self, client, data):
self.__client = client
# Simple object attributes.
self.type = data.pop("album_type", None)
self.group = data.pop("album_group", None)
self.artists = [Artist(client, artist) for artist in data.pop("artists", [])]
self.artist = self.artists[0] if self.artists else None
self.markets = data.pop("avaliable_markets", None)
self.url = data.pop("external_urls").get("spotify", None)
self.id = data.pop("id", None) # pylint: disable=invalid-name
self.name = data.pop("name", None)
self.href = data.pop("href", None)
self.uri = data.pop("uri", None)
self.release_date = data.pop("release_date", None)
self.release_date_precision = data.pop("release_date_precision", None)
self.images = [Image(**image) for image in data.pop("images", [])]
self.restrictions = data.pop("restrictions", None)
# Full object attributes
self.genres = data.pop("genres", None)
self.copyrights = data.pop("copyrights", None)
self.label = data.pop("label", None)
self.popularity = data.pop("popularity", None)
self.total_tracks = data.pop("total_tracks", None)
# AsyncIterable attrs
self.__aiter_klass__ = Track
self.__aiter_fetch__ = partial(
self.__client.http.album_tracks, self.id, limit=50
)
def __repr__(self):
return f"<spotify.Album: {(self.name or self.id or self.uri)!r}>"
# Public
@set_required_scopes(None)
async def get_tracks(
self, *, limit: Optional[int] = 20, offset: Optional[int] = 0
) -> List[Track]:
"""get the albums tracks from spotify.
Parameters
----------
limit : Optional[int]
The limit on how many tracks to retrieve for this album (default is 20).
offset : Optional[int]
The offset from where the api should start from in the tracks.
Returns
-------
tracks : List[Track]
The tracks of the artist.
"""
data = await self.__client.http.album_tracks(
self.id, limit=limit, offset=offset
)
return list(Track(self.__client, item, album=self) for item in data["items"])
@set_required_scopes(None)
async def get_all_tracks(
self, *, market: Optional[str] = "US"
) -> List[Track]: # pylint: disable=unused-argument
"""loads all of the albums tracks, depending on how many the album has this may be a long operation.
Parameters
----------
market : Optional[str]
An ISO 3166-1 alpha-2 country code. Provide this parameter if you want to apply Track Relinking.
Returns
-------
tracks : List[:class:`spotify.Track`]
The tracks of the artist.
"""
return [track async for track in self]

186
pomice/spotify/models/artist.py
View File

@ -1,186 +0,0 @@
from functools import partial
from typing import Optional, List, TYPE_CHECKING
from ..oauth import set_required_scopes
from . import AsyncIterable, URIBase, Image
if TYPE_CHECKING:
import spotify
class Artist(URIBase, AsyncIterable): # pylint: disable=too-many-instance-attributes
"""A Spotify Artist.
Attributes
----------
id : str
The Spotify ID of the artist.
uri : str
The URI of the artist.
url : str
The open.spotify URL.
href : str
A link to the Web API endpoint providing full details of the artist.
name : str
The name of the artist.
genres : List[str]
A list of the genres the artist is associated with.
For example: "Prog Rock" , "Post-Grunge". (If not yet classified, the array is empty.)
followers : Optional[int]
The total number of followers.
popularity : int
The popularity of the artist.
The value will be between 0 and 100, with 100 being the most popular.
The artist’s popularity is calculated from the popularity of all the artist’s tracks.
images : List[Image]
Images of the artist in various sizes, widest first.
"""
def __init__(self, client, data):
self.__client = client
# Simplified object attributes
self.id = data.pop("id") # pylint: disable=invalid-name
self.uri = data.pop("uri")
self.url = data.pop("external_urls").get("spotify", None)
self.href = data.pop("href")
self.name = data.pop("name")
# Full object attributes
self.genres = data.pop("genres", None)
self.followers = data.pop("followers", {}).get("total", None)
self.popularity = data.pop("popularity", None)
self.images = list(Image(**image) for image in data.pop("images", []))
# AsyncIterable attrs
from .album import Album
self.__aiter_klass__ = Album
self.__aiter_fetch__ = partial(
self.__client.http.artist_albums, self.id, limit=50
)
def __repr__(self):
return f"<spotify.Artist: {self.name!r}>"
# Public
@set_required_scopes(None)
async def get_albums(
self,
*,
limit: Optional[int] = 20,
offset: Optional[int] = 0,
include_groups=None,
market: Optional[str] = None,
) -> List["spotify.Album"]:
"""Get the albums of a Spotify artist.
Parameters
----------
limit : Optional[int]
The maximum number of items to return. Default: 20. Minimum: 1. Maximum: 50.
offset : Optiona[int]
The offset of which Spotify should start yielding from.
include_groups : INCLUDE_GROUPS_TP
INCLUDE_GROUPS
market : Optional[str]
An ISO 3166-1 alpha-2 country code.
Returns
-------
albums : List[Album]
The albums of the artist.
"""
from .album import Album
data = await self.__client.http.artist_albums(
self.id,
limit=limit,
offset=offset,
include_groups=include_groups,
market=market,
)
return list(Album(self.__client, item) for item in data["items"])
@set_required_scopes(None)
async def get_all_albums(self, *, market="US") -> List["spotify.Album"]:
"""loads all of the artists albums, depending on how many the artist has this may be a long operation.
Parameters
----------
market : Optional[str]
An ISO 3166-1 alpha-2 country code.
Returns
-------
albums : List[Album]
The albums of the artist.
"""
from .album import Album
albums: List[Album] = []
offset = 0
total = await self.total_albums(market=market)
while len(albums) < total:
data = await self.__client.http.artist_albums(
self.id, limit=50, offset=offset, market=market
)
offset += 50
albums += list(Album(self.__client, item) for item in data["items"])
return albums
@set_required_scopes(None)
async def total_albums(self, *, market: str = None) -> int:
"""get the total amout of tracks in the album.
Parameters
----------
market : Optional[str]
An ISO 3166-1 alpha-2 country code.
Returns
-------
total : int
The total amount of albums.
"""
data = await self.__client.http.artist_albums(
self.id, limit=1, offset=0, market=market
)
return data["total"]
@set_required_scopes(None)
async def top_tracks(self, country: str = "US") -> List["spotify.Track"]:
"""Get Spotify catalog information about an artist’s top tracks by country.
Parameters
----------
country : str
The country to search for, it defaults to 'US'.
Returns
-------
tracks : List[Track]
The artists top tracks.
"""
from .track import Track
top = await self.__client.http.artist_top_tracks(self.id, country=country)
return list(Track(self.__client, item) for item in top["tracks"])
@set_required_scopes(None)
async def related_artists(self) -> List["Artist"]:
"""Get Spotify catalog information about artists similar to a given artist.
Similarity is based on analysis of the Spotify community’s listening history.
Returns
-------
artists : List[Artist]
The artists deemed similar.
"""
related = await self.__client.http.artist_related_artists(self.id)
return list(Artist(self.__client, item) for item in related["artists"])

143
pomice/spotify/models/base.py
View File

@ -1,143 +0,0 @@
from typing import Optional, Callable, Type
import spotify
class SpotifyBase:
"""The base class all Spotify models **must** derive from.
This base class is used to transparently construct spotify
models based on the :class:`spotify,Client` type.
Currently it is used to detect whether a Client is a synchronous
client and, if as such, construct and return the appropriate
synchronous model.
"""
__slots__ = ()
def __new__(cls, client, *_, **__):
if hasattr(client, "__client_thread__"):
cls = getattr( # pylint: disable=self-cls-assignment
spotify.sync.models, cls.__name__
)
return object.__new__(cls)
async def from_href(self):
"""Get the full object from spotify with a `href` attribute.
.. note ::
This can be used to get an updated model of the object.
Returns
-------
model : SpotifyBase
An instance of whatever the class was before.
Raises
------
TypeError
This is raised if the model has no `href` attribute.
Additionally if the model has no `http` attribute and
the model has no way to access its client, while theoretically
impossible its a failsafe, this will be raised.
"""
if not hasattr(self, "href"):
raise TypeError(
"Spotify object has no `href` attribute, therefore cannot be retrived"
)
if hasattr(self, "http"):
return await self.http.request( # pylint: disable=no-member
("GET", self.href) # pylint: disable=no-member
)
klass = type(self)
try:
client = getattr(self, f"_{klass.__name__}__client")
except AttributeError:
raise TypeError("Spotify object has no way to access a HTTPClient.")
else:
http = client.http # pylint: disable=no-member
data = await http.request(("GET", self.href)) # pylint: disable=no-member
return klass(client, data)
class URIBase(SpotifyBase):
"""Base class used for inheriting magic methods for models who have URIs.
This class inherits from :class:`SpotifyBase` and is used to reduce boilerplate
in spotify models by supplying a `__eq__`, `__ne__`, and `__str__` double underscore
methods.
The objects that inherit from :class:`URIBase` support equality and string casting.
- Two objects are equal if **They are strictly the same type and have the same uri**
- Casting to a string will return the uri of the object.
"""
uri = repr(None)
def __hash__(self):
return hash(self.uri) # pylint: disable=no-member
def __eq__(self, other):
return (
type(self) is type(other) and self.uri == other.uri
) # pylint: disable=no-member
def __ne__(self, other):
return not self.__eq__(other)
def __str__(self):
return self.uri # pylint: disable=no-member
class AsyncIterable(SpotifyBase):
"""Base class intended for all models that can be asynchronously iterated over.
This class implements two magic class vars:
* `__aiter_fetch__` ~ A coroutine function that accepts a keyword argument named `option`
* `__aiter_klass__` ~ A spotify model class, essentially a type that subclasses `SpotifyBase`
Additionally the class implements `__aiter__` that will exhaust the paging
objects returned by the `__aiter_fetch__` calls and yield each data item in
said paging objects as an instance of `__aiter_klass__`.
"""
__aiter_fetch__: Optional[Callable] = None
__aiter_klass__: Optional[Type[SpotifyBase]] = None
async def __aiter__(self):
client = getattr(self, f"_{type(self).__name__}__client")
assert self.__aiter_fetch__ is not None
fetch = self.__aiter_fetch__
assert self.__aiter_klass__ is not None
klass = self.__aiter_klass__
total = None
processed = offset = 0
while total is None or processed < total:
data = await fetch(offset=offset) # pylint: disable=not-callable
if total is None:
assert "total" in data
total = data["total"]
assert "items" in data
for item in data["items"]:
processed += 1
yield klass(client, item) # pylint: disable=not-callable
offset += 50

109
pomice/spotify/models/common.py
View File

@ -1,109 +0,0 @@
class Image:
"""An object representing a Spotify image resource.
Attributes
----------
height : :class:`str`
The height of the image.
width : :class:`str`
The width of the image.
url : :class:`str`
The URL of the image.
"""
__slots__ = ("height", "width", "url")
def __init__(self, *, height: str, width: str, url: str):
self.height = height
self.width = width
self.url = url
def __repr__(self):
return f"<spotify.Image: {self.url!r} (width: {self.width!r}, height: {self.height!r})>"
def __eq__(self, other):
return type(self) is type(other) and self.url == other.url
class Context:
"""A Spotify Context.
Attributes
----------
type : str
The object type, e.g. “artist”, “playlist”, “album”.
href : str
A link to the Web API endpoint providing full details of the track.
external_urls : str
External URLs for this context.
uri : str
The Spotify URI for the context.
"""
__slots__ = ("external_urls", "type", "href", "uri")
def __init__(self, data):
self.external_urls = data.get("external_urls")
self.type = data.get("type")
self.href = data.get("href")
self.uri = data.get("uri")
def __repr__(self):
return f"<spotify.Context: {self.uri!r}>"
def __eq__(self, other):
return type(self) is type(other) and self.uri == other.uri
class Device:
"""A Spotify Users device.
Attributes
----------
id : str
The device ID
name : int
The name of the device.
type : str
A Device type, such as “Computer”, “Smartphone” or “Speaker”.
volume : int
The current volume in percent. This may be null.
is_active : bool
if this device is the currently active device.
is_restricted : bool
Whether controlling this device is restricted.
At present if this is “true” then no Web API commands will be accepted by this device.
is_private_session : bool
If this device is currently in a private session.
"""
__slots__ = (
"id",
"name",
"type",
"volume",
"is_active",
"is_restricted",
"is_private_session",
)
def __init__(self, data):
self.id = data.get("id") # pylint: disable=invalid-name
self.name = data.get("name")
self.type = data.get("type")
self.volume = data.get("volume_percent")
self.is_active = data.get("is_active")
self.is_restricted = data.get("is_restricted")
self.is_private_session = data.get("is_private_session")
def __eq__(self, other):
return type(self) is type(other) and self.id == other.id
def __repr__(self):
return f"<spotify.Device: {(self.name or self.id)!r}>"
def __str__(self):
return self.id

189
pomice/spotify/models/library.py
View File

@ -1,189 +0,0 @@
from typing import Sequence, Union, List
from ..oauth import set_required_scopes
from . import SpotifyBase
from .track import Track
from .album import Album
class Library(SpotifyBase):
"""A Spotify Users Library.
Attributes
----------
user : :class:`Spotify.User`
The user which this library object belongs to.
"""
def __init__(self, client, user):
self.user = user
self.__client = client
def __repr__(self):
return f"<spotify.Library: {self.user!r}>"
def __eq__(self, other):
return type(self) is type(other) and self.user == other.user
def __ne__(self, other):
return not self.__eq__(other)
@set_required_scopes("user-library-read")
async def contains_albums(self, *albums: Sequence[Union[str, Album]]) -> List[bool]:
"""Check if one or more albums is already saved in the current Spotify user’s ‘Your Music’ library.
Parameters
----------
albums : Union[Album, str]
A sequence of artist objects or spotify IDs
"""
_albums = [str(obj) for obj in albums]
return await self.user.http.is_saved_album(_albums)
@set_required_scopes("user-library-read")
async def contains_tracks(self, *tracks: Sequence[Union[str, Track]]) -> List[bool]:
"""Check if one or more tracks is already saved in the current Spotify user’s ‘Your Music’ library.
Parameters
----------
tracks : Union[Track, str]
A sequence of track objects or spotify IDs
"""
_tracks = [str(obj) for obj in tracks]
return await self.user.http.is_saved_track(_tracks)
@set_required_scopes("user-library-read")
async def get_tracks(self, *, limit=20, offset=0) -> List[Track]:
"""Get a list of the songs saved in the current Spotify user’s ‘Your Music’ library.
Parameters
----------
limit : Optional[int]
The maximum number of items to return. Default: 20. Minimum: 1. Maximum: 50.
offset : Optional[int]
The index of the first item to return. Default: 0
"""
data = await self.user.http.saved_tracks(limit=limit, offset=offset)
return [Track(self.__client, item["track"]) for item in data["items"]]
@set_required_scopes("user-library-read")
async def get_all_tracks(self) -> List[Track]:
"""Get a list of all the songs saved in the current Spotify user’s ‘Your Music’ library.
Returns
-------
tracks : List[:class:`Track`]
The tracks of the artist.
"""
tracks: List[Track] = []
total = None
offset = 0
while True:
data = await self.user.http.saved_tracks(limit=50, offset=offset)
if total is None:
total = data["total"]
offset += 50
tracks += list(
Track(self.__client, item["track"]) for item in data["items"]
)
if len(tracks) >= total:
break
return tracks
@set_required_scopes("user-library-read")
async def get_albums(self, *, limit=20, offset=0) -> List[Album]:
"""Get a list of the albums saved in the current Spotify user’s ‘Your Music’ library.
Parameters
----------
limit : Optional[int]
The maximum number of items to return. Default: 20. Minimum: 1. Maximum: 50.
offset : Optional[int]
The index of the first item to return. Default: 0
"""
data = await self.user.http.saved_albums(limit=limit, offset=offset)
return [Album(self.__client, item["album"]) for item in data["items"]]
@set_required_scopes("user-library-read")
async def get_all_albums(self) -> List[Album]:
"""Get a list of the albums saved in the current Spotify user’s ‘Your Music’ library.
Returns
-------
albums : List[:class:`Album`]
The albums.
"""
albums: List[Album] = []
total = None
offset = 0
while True:
data = await self.user.http.saved_albums(limit=50, offset=offset)
if total is None:
total = data["total"]
offset += 50
albums += list(
Album(self.__client, item["album"]) for item in data["items"]
)
if len(albums) >= total:
break
return albums
@set_required_scopes("user-library-modify")
async def remove_albums(self, *albums):
"""Remove one or more albums from the current user’s ‘Your Music’ library.
Parameters
----------
albums : Sequence[Union[Album, str]]
A sequence of artist objects or spotify IDs
"""
_albums = [(obj if isinstance(obj, str) else obj.id) for obj in albums]
await self.user.http.delete_saved_albums(",".join(_albums))
@set_required_scopes("user-library-modify")
async def remove_tracks(self, *tracks):
"""Remove one or more tracks from the current user’s ‘Your Music’ library.
Parameters
----------
tracks : Sequence[Union[Track, str]]
A sequence of track objects or spotify IDs
"""
_tracks = [(obj if isinstance(obj, str) else obj.id) for obj in tracks]
await self.user.http.delete_saved_tracks(",".join(_tracks))
@set_required_scopes("user-library-modify")
async def save_albums(self, *albums):
"""Save one or more albums to the current user’s ‘Your Music’ library.
Parameters
----------
albums : Sequence[Union[Album, str]]
A sequence of artist objects or spotify IDs
"""
_albums = [(obj if isinstance(obj, str) else obj.id) for obj in albums]
await self.user.http.save_albums(",".join(_albums))
@set_required_scopes("user-library-modify")
async def save_tracks(self, *tracks):
"""Save one or more tracks to the current user’s ‘Your Music’ library.
Parameters
----------
tracks : Sequence[Union[Track, str]]
A sequence of track objects or spotify IDs
"""
_tracks = [(obj if isinstance(obj, str) else obj.id) for obj in tracks]
await self.user.http.save_tracks(_tracks)

262
pomice/spotify/models/player.py
View File

@ -1,262 +0,0 @@
from typing import Union, Optional, List
from ..oauth import set_required_scopes
from . import SpotifyBase, Device, Track
from .typing import SomeURIs, SomeURI
Offset = Union[int, str, Track]
SomeDevice = Union[Device, str]
class Player(SpotifyBase): # pylint: disable=too-many-instance-attributes
"""A Spotify Users current playback.
Attributes
----------
device : :class:`spotify.Device`
The device that is currently active.
repeat_state : :class:`str`
"off", "track", "context"
shuffle_state : :class:`bool`
If shuffle is on or off.
is_playing : :class:`bool`
If something is currently playing.
"""
def __init__(self, client, user, data):
self.__client = client
self.__user = user
self.repeat_state = data.get("repeat_state", None)
self.shuffle_state = data.pop("shuffle_state", None)
self.is_playing = data.pop("is_playing", None)
self.device = Device(data=data.pop("device", None))
def __repr__(self):
return f"<spotify.Player: {self.user!r}>"
# Properties
@property
def user(self):
return self.__user
# Public methods
@set_required_scopes("user-modify-playback-state")
async def pause(self, *, device: Optional[SomeDevice] = None):
"""Pause playback on the user’s account.
Parameters
----------
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.pause_playback(device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def resume(self, *, device: Optional[SomeDevice] = None):
"""Resume playback on the user's account.
Parameters
----------
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.play_playback(None, device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def seek(self, pos, *, device: Optional[SomeDevice] = None):
"""Seeks to the given position in the user’s currently playing track.
Parameters
----------
pos : int
The position in milliseconds to seek to.
Must be a positive number.
Passing in a position that is greater than the length of the track will cause the player to start playing the next song.
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.seek_playback(pos, device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def set_repeat(self, state, *, device: Optional[SomeDevice] = None):
"""Set the repeat mode for the user’s playback.
Parameters
----------
state : str
Options are repeat-track, repeat-context, and off
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.repeat_playback(state, device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def set_volume(self, volume: int, *, device: Optional[SomeDevice] = None):
"""Set the volume for the user’s current playback device.
Parameters
----------
volume : int
The volume to set. Must be a value from 0 to 100 inclusive.
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.set_playback_volume(volume, device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def next(self, *, device: Optional[SomeDevice] = None):
"""Skips to next track in the user’s queue.
Parameters
----------
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.skip_next(device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def previous(self, *, device: Optional[SomeDevice] = None):
"""Skips to previous track in the user’s queue.
Note that this will ALWAYS skip to the previous track, regardless of the current track’s progress.
Returning to the start of the current track should be performed using :meth:`seek`
Parameters
----------
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
return await self.user.http.skip_previous(device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def enqueue(self, uri: SomeURI, device: Optional[SomeDevice] = None):
"""Add an item to the end of the user’s current playback queue.
Parameters
----------
uri : Union[:class:`spotify.URIBase`, :class:`str`]
The uri of the item to add to the queue. Must be a track or an
episode uri.
device_id : Optional[Union[Device, :class:`str`]]
The id of the device this command is targeting. If not supplied,
the user’s currently active device is the target.
"""
device_id: Optional[str]
if device is not None:
if not isinstance(device, (Device, str)):
raise TypeError(
f"Expected `device` to either be a spotify.Device or a string. got {type(device)!r}"
)
device_id = str(device)
else:
device_id = None
await self.user.http.playback_queue(uri=str(uri), device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def play(
self,
*uris: SomeURIs,
offset: Optional[Offset] = 0,
device: Optional[SomeDevice] = None,
):
"""Start a new context or resume current playback on the user’s active device.
The method treats a single argument as a Spotify context, such as a Artist, Album and playlist objects/URI.
When called with multiple positional arguments they are interpreted as a array of Spotify Track objects/URIs.
Parameters
----------
*uris : SomeURI
When a single argument is passed in that argument is treated
as a context (except if it is a track or track uri).
Valid contexts are: albums, artists, playlists.
Album, Artist and Playlist objects are accepted too.
Otherwise when multiple arguments are passed in they,
A sequence of Spotify Tracks or Track URIs to play.
offset : Optional[:obj:`Offset`]
Indicates from where in the context playback should start.
Only available when `context` corresponds to an album or playlist object,
or when the `uris` parameter is used. when an integer offset is zero based and can’t be negative.
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
context_uri: Union[List[str], str]
if (
len(uris) > 1
or isinstance(uris[0], Track)
or (isinstance(uris[0], str) and "track" in uris[0])
):
# Regular uris paramter
context_uri = [str(uri) for uri in uris]
else:
# Treat it as a context URI
context_uri = str(uris[0])
device_id: Optional[str]
if device is not None:
if not isinstance(device, (Device, str)):
raise TypeError(
f"Expected `device` to either be a spotify.Device or a string. got {type(device)!r}"
)
device_id = str(device)
else:
device_id = None
await self.user.http.play_playback(
context_uri, offset=offset, device_id=device_id
)
@set_required_scopes("user-modify-playback-state")
async def shuffle(
self, state: Optional[bool] = None, *, device: Optional[SomeDevice] = None
):
"""shuffle on or off for user’s playback.
Parameters
----------
state : Optional[bool]
if `True` then Shuffle user’s playback.
else if `False` do not shuffle user’s playback.
device : Optional[:obj:`SomeDevice`]
The Device object or id of the device this command is targeting.
If not supplied, the user’s currently active device is the target.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.shuffle_playback(state, device_id=device_id)
@set_required_scopes("user-modify-playback-state")
async def transfer(self, device: SomeDevice, ensure_playback: bool = False):
"""Transfer playback to a new device and determine if it should start playing.
Parameters
----------
device : :obj:`SomeDevice`
The device on which playback should be started/transferred.
ensure_playback : bool
if `True` ensure playback happens on new device.
else keep the current playback state.
"""
device_id: Optional[str] = str(device) if device is not None else None
await self.user.http.transfer_player(device_id=device_id, play=ensure_playback)

525
pomice/spotify/models/playlist.py
View File

@ -1,525 +0,0 @@
from functools import partial
from itertools import islice
from typing import List, Optional, Union, Callable, Tuple, Iterable, TYPE_CHECKING, Any, Dict, Set
from ..oauth import set_required_scopes
from ..http import HTTPUserClient, HTTPClient
from . import AsyncIterable, URIBase, Track, PlaylistTrack, Image
if TYPE_CHECKING:
import spotify
class MutableTracks:
__slots__ = (
"playlist",
"tracks",
"was_empty",
"is_empty",
"replace_tracks",
"get_all_tracks",
)
def __init__(self, playlist: "Playlist") -> None:
self.playlist = playlist
self.tracks = tracks = getattr(playlist, "_Playlist__tracks")
if tracks is not None:
self.was_empty = self.is_empty = not tracks
self.replace_tracks = playlist.replace_tracks
self.get_all_tracks = playlist.get_all_tracks
async def __aenter__(self):
if self.tracks is None:
self.tracks = tracks = list(await self.get_all_tracks())
self.was_empty = self.is_empty = not tracks
else:
tracks = list(self.tracks)
return tracks
async def __aexit__(self, typ, value, traceback):
if self.was_empty and self.is_empty:
# the tracks were empty and is still empty.
# skip the api call.
return
tracks = self.tracks
await self.replace_tracks(*tracks)
setattr(self.playlist, "_Playlist__tracks", tuple(self.tracks))
class Playlist(URIBase, AsyncIterable): # pylint: disable=too-many-instance-attributes
"""A Spotify Playlist.
Attributes
----------
collaborative : :class:`bool`
Returns true if context is not search and the owner allows other users to modify the playlist. Otherwise returns false.
description : :class:`str`
The playlist description. Only returned for modified, verified playlists, otherwise null.
url : :class:`str`
The open.spotify URL.
followers : :class:`int`
The total amount of followers
href : :class:`str`
A link to the Web API endpoint providing full details of the playlist.
id : :class:`str`
The Spotify ID for the playlist.
images : List[:class:`spotify.Image`]
Images for the playlist.
The array may be empty or contain up to three images.
The images are returned by size in descending order.
If returned, the source URL for the image ( url ) is temporary and will expire in less than a day.
name : :class:`str`
The name of the playlist.
owner : :class:`spotify.User`
The user who owns the playlist
public : :class`bool`
The playlist’s public/private status:
true the playlist is public,
false the playlist is private,
null the playlist status is not relevant.
snapshot_id : :class:`str`
The version identifier for the current playlist.
tracks : Optional[Tuple[:class:`PlaylistTrack`]]
A tuple of :class:`PlaylistTrack` objects or `None`.
"""
__slots__ = (
"collaborative",
"description",
"url",
"followers",
"href",
"id",
"images",
"name",
"owner",
"public",
"uri",
"total_tracks",
"__client",
"__http",
"__tracks",
)
__tracks: Optional[Tuple[PlaylistTrack, ...]]
__http: Union[HTTPUserClient, HTTPClient]
total_tracks: Optional[int]
def __init__(
self,
client: "spotify.Client",
data: Union[dict, "Playlist"],
*,
http: Optional[HTTPClient] = None,
):
self.__client = client
self.__http = http or client.http
assert self.__http is not None
self.__tracks = None
self.total_tracks = None
if not isinstance(data, (Playlist, dict)):
raise TypeError("data must be a Playlist instance or a dict.")
if isinstance(data, dict):
self.__from_raw(data)
else:
for name in filter((lambda name: name[0] != "_"), Playlist.__slots__):
setattr(self, name, getattr(data, name))
# AsyncIterable attrs
self.__aiter_klass__ = PlaylistTrack
self.__aiter_fetch__ = partial(
client.http.get_playlist_tracks, self.id, limit=50
)
def __repr__(self):
return f'<spotify.Playlist: {getattr(self, "name", None) or self.id}>'
def __len__(self):
return self.total_tracks
# Internals
def __from_raw(self, data: dict) -> None:
from .user import User
client = self.__client
self.id = data.pop("id") # pylint: disable=invalid-name
self.images = tuple(Image(**image) for image in data.pop("images", []))
self.owner = User(client, data=data.pop("owner"))
self.public = data.pop("public")
self.collaborative = data.pop("collaborative")
self.description = data.pop("description", None)
self.followers = data.pop("followers", {}).get("total", None)
self.href = data.pop("href")
self.name = data.pop("name")
self.url = data.pop("external_urls").get("spotify", None)
self.uri = data.pop("uri")
tracks: Optional[Tuple[PlaylistTrack, ...]] = (
tuple(PlaylistTrack(client, item) for item in data["tracks"]["items"])
if "items" in data["tracks"]
else None
)
self.__tracks = tracks
self.total_tracks = (
data["tracks"]["total"]
)
# Track retrieval
@set_required_scopes(None)
async def get_tracks(
self, *, limit: Optional[int] = 20, offset: Optional[int] = 0
) -> Tuple[PlaylistTrack, ...]:
"""Get a fraction of a playlists tracks.
Parameters
----------
limit : Optional[int]
The limit on how many tracks to retrieve for this playlist (default is 20).
offset : Optional[int]
The offset from where the api should start from in the tracks.
Returns
-------
tracks : Tuple[PlaylistTrack]
The tracks of the playlist.
"""
data = await self.__http.get_playlist_tracks(
self.id, limit=limit, offset=offset
)
return tuple(PlaylistTrack(self.__client, item) for item in data["items"])
@set_required_scopes(None)
async def get_all_tracks(self) -> Tuple[PlaylistTrack, ...]:
"""Get all playlist tracks from the playlist.
Returns
-------
tracks : Tuple[:class:`PlaylistTrack`]
The playlists tracks.
"""
tracks: List[PlaylistTrack] = []
offset = 0
if self.total_tracks is None:
self.total_tracks = (
await self.__http.get_playlist_tracks(self.id, limit=1, offset=0)
)["total"]
while len(tracks) < self.total_tracks:
data = await self.__http.get_playlist_tracks(
self.id, limit=50, offset=offset
)
tracks += [PlaylistTrack(self.__client, item) for item in data["items"]]
offset += 50
self.total_tracks = len(tracks)
return tuple(tracks)
# Playlist structure modification
# Basic api wrapping
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def add_tracks(self, *tracks) -> str:
"""Add one or more tracks to a user’s playlist.
Parameters
----------
tracks : Iterable[Union[:class:`str`, :class:`Track`]]
Tracks to add to the playlist
Returns
-------
snapshot_id : :class:`str`
The snapshot id of the playlist.
"""
data = await self.__http.add_playlist_tracks(
self.id, tracks=[str(track) for track in tracks]
)
return data["snapshot_id"]
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def remove_tracks(
self, *tracks: Union[str, Track, Tuple[Union[str, Track], List[int]]]
):
"""Remove one or more tracks from a user’s playlist.
Parameters
----------
tracks : Iterable[Union[:class:`str`, :class:`Track`]]
Tracks to remove from the playlist
Returns
-------
snapshot_id : :class:`str`
The snapshot id of the playlist.
"""
tracks_: List[Union[str, Dict[str, Union[str, Set[int]]]]] = []
for part in tracks:
if not isinstance(part, (Track, str, tuple)):
raise TypeError(
"Track argument of tracks parameter must be a Track instance, string or a tuple of those and an iterator of positive integers."
)
if isinstance(part, (Track, str)):
tracks_.append(str(part))
continue
track, positions, = part
if not isinstance(track, (Track, str)):
raise TypeError(
"Track argument of tuple track parameter must be a Track instance or a string."
)
if not hasattr(positions, "__iter__"):
raise TypeError("Positions element of track tuple must be a iterator.")
if not all(isinstance(index, int) for index in positions):
raise TypeError("Members of the positions iterator must be integers.")
elem: Dict[str, Union[str, Set[int]]] = {"uri": str(track), "positions": set(positions)}
tracks_.append(elem)
data = await self.__http.remove_playlist_tracks(self.id, tracks=tracks_)
return data["snapshot_id"]
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def replace_tracks(self, *tracks: Union[Track, PlaylistTrack, str]) -> None:
"""Replace all the tracks in a playlist, overwriting its existing tracks.
This powerful request can be useful for replacing tracks, re-ordering existing tracks, or clearing the playlist.
Parameters
----------
tracks : Iterable[Union[:class:`str`, :class:`Track`]]
Tracks to place in the playlist
"""
bucket: List[str] = []
for track in tracks:
if not isinstance(track, (str, Track)):
raise TypeError(
f"tracks must be a iterable of strings or Track instances. Got {type(track)!r}"
)
bucket.append(str(track))
body: Tuple[str, ...] = tuple(bucket)
head: Tuple[str, ...]
tail: Tuple[str, ...]
head, tail = body[:100], body[100:]
if head:
await self.__http.replace_playlist_tracks(self.id, tracks=head)
while tail:
head, tail = tail[:100], tail[100:]
await self.extend(head)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def reorder_tracks(
self,
start: int,
insert_before: int,
length: int = 1,
*,
snapshot_id: Optional[str] = None,
) -> str:
"""Reorder a track or a group of tracks in a playlist.
Parameters
----------
start : int
The position of the first track to be reordered.
insert_before : int
The position where the tracks should be inserted.
length : Optional[int]
The amount of tracks to be reordered. Defaults to 1 if not set.
snapshot_id : str
The playlist’s snapshot ID against which you want to make the changes.
Returns
-------
snapshot_id : str
The snapshot id of the playlist.
"""
data = await self.__http.reorder_playlists_tracks(
self.id, start, length, insert_before, snapshot_id=snapshot_id
)
return data["snapshot_id"]
# Library functionality.
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def clear(self):
"""Clear the playlists tracks.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
.. warning::
This is a desctructive operation and can not be reversed!
"""
await self.__http.replace_playlist_tracks(self.id, tracks=[])
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def extend(self, tracks: Union["Playlist", Iterable[Union[Track, str]]]):
"""Extend a playlists tracks with that of another playlist or a list of Track/Track URIs.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
Parameters
----------
tracks : Union["Playlist", List[Union[Track, str]]]
Tracks to add to the playlist, acceptable values are:
- A :class:`spotify.Playlist` object
- A :class:`list` of :class:`spotify.Track` objects or Track URIs
Returns
-------
snapshot_id : str
The snapshot id of the playlist.
"""
bucket: Iterable[Union[Track, str]]
if isinstance(tracks, Playlist):
bucket = await tracks.get_all_tracks()
elif not hasattr(tracks, "__iter__"):
raise TypeError(
f"`tracks` was an invalid type, expected any of: Playlist, Iterable[Union[Track, str]], instead got {type(tracks)}"
)
else:
bucket = list(tracks)
gen: Iterable[str] = (str(track) for track in bucket)
while True:
head: List[str] = list(islice(gen, 0, 100))
if not head:
break
await self.__http.add_playlist_tracks(self.id, tracks=head)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def insert(self, index, obj: Union[PlaylistTrack, Track]) -> None:
"""Insert an object before the index.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
"""
if not isinstance(obj, (PlaylistTrack, Track)):
raise TypeError(
f"Expected a PlaylistTrack or Track object instead got {obj!r}"
)
async with MutableTracks(self) as tracks:
tracks.insert(index, obj)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def pop(self, index: int = -1) -> PlaylistTrack:
"""Remove and return the track at the specified index.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
Returns
-------
playlist_track : :class:`PlaylistTrack`
The track that was removed.
Raises
------
IndexError
If there are no tracks or the index is out of range.
"""
async with MutableTracks(self) as tracks:
return tracks.pop(index)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def sort(
self,
*,
key: Optional[Callable[[PlaylistTrack], bool]] = None,
reverse: Optional[bool] = False,
) -> None:
"""Stable sort the playlist in place.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
"""
async with MutableTracks(self) as tracks:
tracks.sort(key=key, reverse=reverse)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def remove(self, value: Union[PlaylistTrack, Track]) -> None:
"""Remove the first occurence of the value.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
Raises
-------
ValueError
If the value is not present.
"""
async with MutableTracks(self) as tracks:
tracks.remove(value)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def copy(self) -> "Playlist":
"""Return a shallow copy of the playlist object.
Returns
-------
playlist : :class:`Playlist`
The playlist object copy.
"""
return Playlist(client=self.__client, data=self, http=self.__http)
@set_required_scopes("playlist-modify-public", "playlist-modify-private")
async def reverse(self) -> None:
"""Reverse the playlist in place.
.. note::
This method will mutate the current
playlist object, and the spotify Playlist.
"""
async with MutableTracks(self) as tracks:
tracks.reverse()

123
pomice/spotify/models/track.py
View File

@ -1,123 +0,0 @@
"""Source implementation for spotify Tracks, and any other semantically relevent, implementation."""
import datetime
from itertools import starmap
from ..oauth import set_required_scopes
from . import URIBase, Image, Artist
class Track(URIBase): # pylint: disable=too-many-instance-attributes
"""A Spotify Track object.
Attribtues
----------
id : :class:`str`
The Spotify ID for the track.
name : :class:`str`
The name of the track.
href : :class:`str`
A link to the Web API endpoint providing full details of the track.
uri : :class:`str`
The Spotify URI for the track.
duration : int
The track length in milliseconds.
explicit : bool
Whether or not the track has explicit
`True` if it does.
`False` if it does not (or unknown)
disc_number : int
The disc number (usually 1 unless the album consists of more than one disc).
track_number : int
The number of the track.
If an album has several discs, the track number is the number on the specified disc.
url : :class:`str`
The open.spotify URL for this Track
is_local : bool
Whether or not the track is from a local file.
popularity : int
POPULARITY
preview_url : :class:`str`
The preview URL for this Track.
images : List[Image]
The images of the Track.
markets : List[:class:`str`]
The available markets for the Track.
"""
def __init__(self, client, data, album=None):
from .album import Album
self.__client = client
self.artists = artists = list(
Artist(client, artist) for artist in data.pop("artists", [])
)
self.artist = artists[-1] if artists else None
album_ = data.pop("album", None)
self.album = Album(client, album_) if album_ else album
self.id = data.pop("id", None) # pylint: disable=invalid-name
self.name = data.pop("name", None)
self.href = data.pop("href", None)
self.uri = data.pop("uri", None)
self.duration = data.pop("duration_ms", None)
self.explicit = data.pop("explicit", None)
self.disc_number = data.pop("disc_number", None)
self.track_number = data.pop("track_number", None)
self.url = data.pop("external_urls").get("spotify", None)
self.is_local = data.pop("is_local", None)
self.popularity = data.pop("popularity", None)
self.preview_url = data.pop("preview_url", None)
self.markets = data.pop("available_markets", [])
if "images" in data:
self.images = list(starmap(Image, data.pop("images")))
else:
self.images = self.album.images.copy() if self.album is not None else []
def __repr__(self):
return f"<spotify.Track: {self.name!r}>"
@set_required_scopes(None)
def audio_analysis(self):
"""Get a detailed audio analysis for the track."""
return self.__client.http.track_audio_analysis(self.id)
@set_required_scopes(None)
def audio_features(self):
"""Get audio feature information for the track."""
return self.__client.http.track_audio_features(self.id)
class PlaylistTrack(Track, URIBase):
"""A Track on a Playlist.
Like a regular :class:`Track` but has some additional attributes.
Attributes
----------
added_by : :class:`str`
The Spotify user who added the track.
is_local : bool
Whether this track is a local file or not.
added_at : datetime.datetime
The datetime of when the track was added to the playlist.
"""
__slots__ = ("added_at", "added_by", "is_local")
def __init__(self, client, data):
from .user import User
super().__init__(client, data["track"])
self.added_by = User(client, data["added_by"])
self.added_at = datetime.datetime.strptime(
data["added_at"], "%Y-%m-%dT%H:%M:%SZ"
)
self.is_local = data["is_local"]
def __repr__(self):
return f"<spotify.PlaylistTrack: {self.name!r}>"

9
pomice/spotify/models/typing.py
View File

@ -1,9 +0,0 @@
"""Type annotation aliases for other Spotify models."""
from typing import Union, Sequence
from .base import URIBase
SomeURI = Union[URIBase, str]
SomeURIs = Sequence[Union[URIBase, str]]
OneOrMoreURIs = Union[SomeURI, Sequence[SomeURI]]

562
pomice/spotify/models/user.py
View File

@ -1,562 +0,0 @@
"""Source implementation for a spotify User"""
import functools
from functools import partial
from base64 import b64encode
from typing import (
Optional,
Dict,
Union,
List,
Type,
TypeVar,
TYPE_CHECKING,
)
from ..utils import to_id
from ..http import HTTPUserClient
from . import (
AsyncIterable,
URIBase,
Image,
Device,
Context,
Player,
Playlist,
Track,
Artist,
Library,
)
if TYPE_CHECKING:
import spotify
T = TypeVar("T", Artist, Track) # pylint: disable=invalid-name
def ensure_http(func):
func.__ensure_http__ = True
return func
class User(URIBase, AsyncIterable): # pylint: disable=too-many-instance-attributes
"""A Spotify User.
Attributes
----------
id : :class:`str`
The Spotify user ID for the user.
uri : :class:`str`
The Spotify URI for the user.
url : :class:`str`
The open.spotify URL.
href : :class:`str`
A link to the Web API endpoint for this user.
display_name : :class:`str`
The name displayed on the user’s profile.
`None` if not available.
followers : :class:`int`
The total number of followers.
images : List[:class:`Image`]
The user’s profile image.
email : :class:`str`
The user’s email address, as entered by the user when creating their account.
country : :class:`str`
The country of the user, as set in the user’s account profile. An ISO 3166-1 alpha-2 country code.
birthdate : :class:`str`
The user’s date-of-birth.
product : :class:`str`
The user’s Spotify subscription level: “premium”, “free”, etc.
(The subscription level “open” can be considered the same as “free”.)
"""
def __init__(self, client: "spotify.Client", data: dict, **kwargs):
self.__client = self.client = client
if "http" not in kwargs:
self.library = None
self.http = client.http
else:
self.http = kwargs.pop("http")
self.library = Library(client, self)
# Public user object attributes
self.id = data.pop("id") # pylint: disable=invalid-name
self.uri = data.pop("uri")
self.url = data.pop("external_urls").get("spotify", None)
self.display_name = data.pop("display_name", None)
self.href = data.pop("href")
self.followers = data.pop("followers", {}).get("total", None)
self.images = list(Image(**image) for image in data.pop("images", []))
# Private user object attributes
self.email = data.pop("email", None)
self.country = data.pop("country", None)
self.birthdate = data.pop("birthdate", None)
self.product = data.pop("product", None)
# AsyncIterable attrs
self.__aiter_klass__ = Playlist
self.__aiter_fetch__ = partial(
self.__client.http.get_playlists, self.id, limit=50
)
def __repr__(self):
return f"<spotify.User: {(self.display_name or self.id)!r}>"
def __getattr__(self, attr):
value = object.__getattribute__(self, attr)
if (
hasattr(value, "__ensure_http__")
and getattr(self, "http", None) is not None
):
@functools.wraps(value)
def _raise(*args, **kwargs):
raise AttributeError(
"User has not HTTP presence to perform API requests."
)
return _raise
return value
async def __aenter__(self) -> "User":
return self
async def __aexit__(self, _, __, ___):
await self.http.close()
# Internals
async def _get_top(self, klass: Type[T], kwargs: dict) -> List[T]:
target = {Artist: "artists", Track: "tracks"}[klass]
data = {
key: value
for key, value in kwargs.items()
if key in ("limit", "offset", "time_range")
}
resp = await self.http.top_artists_or_tracks(target, **data) # type: ignore
return [klass(self.__client, item) for item in resp["items"]]
### Alternate constructors
@classmethod
async def from_code(
cls, client: "spotify.Client", code: str, *, redirect_uri: str,
):
"""Create a :class:`User` object from an authorization code.
Parameters
----------
client : :class:`spotify.Client`
The spotify client to associate the user with.
code : :class:`str`
The authorization code to use to further authenticate the user.
redirect_uri : :class:`str`
The rediriect URI to use in tandem with the authorization code.
"""
route = ("POST", "https://accounts.spotify.com/api/token")
payload = {
"redirect_uri": redirect_uri,
"grant_type": "authorization_code",
"code": code,
}
client_id = client.http.client_id
client_secret = client.http.client_secret
headers = {
"Authorization": f"Basic {b64encode(':'.join((client_id, client_secret)).encode()).decode()}",
"Content-Type": "application/x-www-form-urlencoded",
}
raw = await client.http.request(route, headers=headers, params=payload)
token = raw["access_token"]
refresh_token = raw["refresh_token"]
return await cls.from_token(client, token, refresh_token)
@classmethod
async def from_token(
cls,
client: "spotify.Client",
token: Optional[str],
refresh_token: Optional[str] = None,
):
"""Create a :class:`User` object from an access token.
Parameters
----------
client : :class:`spotify.Client`
The spotify client to associate the user with.
token : :class:`str`
The access token to use for http requests.
refresh_token : :class:`str`
Used to acquire new token when it expires.
"""
client_id = client.http.client_id
client_secret = client.http.client_secret
http = HTTPUserClient(client_id, client_secret, token, refresh_token)
data = await http.current_user()
return cls(client, data=data, http=http)
@classmethod
async def from_refresh_token(cls, client: "spotify.Client", refresh_token: str):
"""Create a :class:`User` object from a refresh token.
It will poll the spotify API for a new access token and
use that to initialize the spotify user.
Parameters
----------
client : :class:`spotify.Client`
The spotify client to associate the user with.
refresh_token: str
Used to acquire token.
"""
return await cls.from_token(client, None, refresh_token)
### Contextual methods
@ensure_http
async def currently_playing(self) -> Dict[str, Union[Track, Context, str]]:
"""Get the users currently playing track.
Returns
-------
context, track : Dict[str, Union[Track, Context, str]]
A tuple of the context and track.
"""
data = await self.http.currently_playing() # type: ignore
if "item" in data:
context = data.pop("context", None)
if context is not None:
data["context"] = Context(context)
else:
data["context"] = None
data["item"] = Track(self.__client, data.get("item", {}) or {})
return data
@ensure_http
async def get_player(self) -> Player:
"""Get information about the users current playback.
Returns
-------
player : :class:`Player`
A player object representing the current playback.
"""
player = Player(self.__client, self, await self.http.current_player()) # type: ignore
return player
@ensure_http
async def get_devices(self) -> List[Device]:
"""Get information about the users avaliable devices.
Returns
-------
devices : List[:class:`Device`]
The devices the user has available.
"""
data = await self.http.available_devices() # type: ignore
return [Device(item) for item in data["devices"]]
@ensure_http
async def recently_played(
self,
*,
limit: int = 20,
before: Optional[str] = None,
after: Optional[str] = None,
) -> List[Dict[str, Union[Track, Context, str]]]:
"""Get tracks from the current users recently played tracks.
Returns
-------
playlist_history : List[Dict[:class:`str`, Union[Track, Context, :class:`str`]]]
A list of playlist history object.
Each object is a dict with a timestamp, track and context field.
"""
data = await self.http.recently_played(limit=limit, before=before, after=after) # type: ignore
client = self.__client
# List[T] where T: {'track': Track, 'content': Context: 'timestamp': ISO8601}
return [
{
"played_at": track.get("played_at"),
"context": Context(track.get("context", {}) or {}),
"track": Track(client, track.get("track", {}) or {}),
}
for track in data["items"]
]
### Playlist track methods
@ensure_http
async def add_tracks(self, playlist: Union[str, Playlist], *tracks) -> str:
"""Add one or more tracks to a user’s playlist.
Parameters
----------
playlist : Union[:class:`str`, Playlist]
The playlist to modify
tracks : Sequence[Union[:class:`str`, Track]]
Tracks to add to the playlist
Returns
-------
snapshot_id : :class:`str`
The snapshot id of the playlist.
"""
data = await self.http.add_playlist_tracks( # type: ignore
to_id(str(playlist)), tracks=[str(track) for track in tracks]
)
return data["snapshot_id"]
@ensure_http
async def replace_tracks(self, playlist, *tracks) -> None:
"""Replace all the tracks in a playlist, overwriting its existing tracks.
This powerful request can be useful for replacing tracks, re-ordering existing tracks, or clearing the playlist.
Parameters
----------
playlist : Union[:class:`str`, PLaylist]
The playlist to modify
tracks : Sequence[Union[:class:`str`, Track]]
Tracks to place in the playlist
"""
await self.http.replace_playlist_tracks( # type: ignore
to_id(str(playlist)), tracks=",".join(str(track) for track in tracks)
)
@ensure_http
async def remove_tracks(self, playlist, *tracks):
"""Remove one or more tracks from a user’s playlist.
Parameters
----------
playlist : Union[:class:`str`, Playlist]
The playlist to modify
tracks : Sequence[Union[:class:`str`, Track]]
Tracks to remove from the playlist
Returns
-------
snapshot_id : :class:`str`
The snapshot id of the playlist.
"""
data = await self.http.remove_playlist_tracks( # type: ignore
to_id(str(playlist)), tracks=(str(track) for track in tracks)
)
return data["snapshot_id"]
@ensure_http
async def reorder_tracks(
self, playlist, start, insert_before, length=1, *, snapshot_id=None
):
"""Reorder a track or a group of tracks in a playlist.
Parameters
----------
playlist : Union[:class:`str`, Playlist]
The playlist to modify
start : int
The position of the first track to be reordered.
insert_before : int
The position where the tracks should be inserted.
length : Optional[int]
The amount of tracks to be reordered. Defaults to 1 if not set.
snapshot_id : :class:`str`
The playlist’s snapshot ID against which you want to make the changes.
Returns
-------
snapshot_id : :class:`str`
The snapshot id of the playlist.
"""
data = await self.http.reorder_playlists_tracks( # type: ignore
to_id(str(playlist)), start, length, insert_before, snapshot_id=snapshot_id
)
return data["snapshot_id"]
### Playlist methods
@ensure_http
async def edit_playlist(
self, playlist, *, name=None, public=None, collaborative=None, description=None
):
"""Change a playlist’s name and public/private, collaborative state and description.
Parameters
----------
playlist : Union[:class:`str`, Playlist]
The playlist to modify
name : Optional[:class:`str`]
The new name of the playlist.
public : Optional[bool]
The public/private status of the playlist.
`True` for public, `False` for private.
collaborative : Optional[bool]
If `True`, the playlist will become collaborative and other users will be able to modify the playlist.
description : Optional[:class:`str`]
The new playlist description
"""
kwargs = {
"name": name,
"public": public,
"collaborative": collaborative,
"description": description,
}
await self.http.change_playlist_details(to_id(str(playlist)), **kwargs) # type: ignore
@ensure_http
async def create_playlist(
self, name, *, public=True, collaborative=False, description=None
):
"""Create a playlist for a Spotify user.
Parameters
----------
name : :class:`str`
The name of the playlist.
public : Optional[bool]
The public/private status of the playlist.
`True` for public, `False` for private.
collaborative : Optional[bool]
If `True`, the playlist will become collaborative and other users will be able to modify the playlist.
description : Optional[:class:`str`]
The playlist description
Returns
-------
playlist : :class:`Playlist`
The playlist that was created.
"""
data = {"name": name, "public": public, "collaborative": collaborative}
if description:
data["description"] = description
playlist_data = await self.http.create_playlist(self.id, **data) # type: ignore
return Playlist(self.__client, playlist_data, http=self.http)
@ensure_http
async def follow_playlist(
self, playlist: Union[str, Playlist], *, public: bool = True
) -> None:
"""follow a playlist
Parameters
----------
playlist : Union[:class:`str`, Playlist]
The playlist to modify
public : Optional[bool]
The public/private status of the playlist.
`True` for public, `False` for private.
"""
await self.http.follow_playlist(to_id(str(playlist)), public=public) # type: ignore
@ensure_http
async def get_playlists(
self, *, limit: int = 20, offset: int = 0
) -> List[Playlist]:
"""get the users playlists from spotify.
Parameters
----------
limit : Optional[int]
The limit on how many playlists to retrieve for this user (default is 20).
offset : Optional[int]
The offset from where the api should start from in the playlists.
Returns
-------
playlists : List[Playlist]
A list of the users playlists.
"""
data = await self.http.get_playlists(self.id, limit=limit, offset=offset) # type: ignore
return [
Playlist(self.__client, playlist_data, http=self.http)
for playlist_data in data["items"]
]
@ensure_http
async def get_all_playlists(self) -> List[Playlist]:
"""Get all of the users playlists from spotify.
Returns
-------
playlists : List[:class:`Playlist`]
A list of the users playlists.
"""
playlists: List[Playlist] = []
total = None
offset = 0
while True:
data = await self.http.get_playlists(self.id, limit=50, offset=offset) # type: ignore
if total is None:
total = data["total"]
offset += 50
playlists += [
Playlist(self.__client, playlist_data, http=self.http)
for playlist_data in data["items"]
]
if len(playlists) >= total:
break
return playlists
@ensure_http
async def top_artists(self, **data) -> List[Artist]:
"""Get the current user’s top artists based on calculated affinity.
Parameters
----------
limit : Optional[int]
The number of entities to return. Default: 20. Minimum: 1. Maximum: 50.
offset : Optional[int]
The index of the first entity to return. Default: 0
time_range : Optional[:class:`str`]
Over what time frame the affinities are computed. (long_term, short_term, medium_term)
Returns
-------
tracks : List[Artist]
The top artists for the user.
"""
return await self._get_top(Artist, data)
@ensure_http
async def top_tracks(self, **data) -> List[Track]:
"""Get the current user’s top tracks based on calculated affinity.
Parameters
----------
limit : Optional[int]
The number of entities to return. Default: 20. Minimum: 1. Maximum: 50.
offset : Optional[int]
The index of the first entity to return. Default: 0
time_range : Optional[:class:`str`]
Over what time frame the affinities are computed. (long_term, short_term, medium_term)
Returns
-------
tracks : List[Track]
The top tracks for the user.
"""
return await self._get_top(Track, data)

202
pomice/spotify/oauth.py
View File

@ -1,202 +0,0 @@
from urllib.parse import quote_plus as quote
from typing import Optional, Dict, Iterable, Union, Set, Callable, Tuple, Any
VALID_SCOPES = (
# Playlists
"playlist-read-collaborative"
"playlist-modify-private"
"playlist-modify-public"
"playlist-read-private"
# Spotify Connect
"user-modify-playback-state"
"user-read-currently-playing"
"user-read-playback-state"
# Users
"user-read-private"
"user-read-email"
# Library
"user-library-modify"
"user-library-read"
# Follow
"user-follow-modify"
"user-follow-read"
# Listening History
"user-read-recently-played"
"user-top-read"
# Playback
"streaming"
"app-remote-control"
)
def set_required_scopes(*scopes: Optional[str]) -> Callable:
"""A decorator that lets you attach metadata to functions.
Parameters
----------
scopes : :class:`str`
A series of scopes that are required.
Returns
-------
decorator : :class:`typing.Callable`
The decorator that sets the scope metadata.
"""
def decorate(func) -> Callable:
func.__requires_spotify_scopes__ = tuple(scopes)
return func
return decorate
def get_required_scopes(func: Callable[..., Any]) -> Tuple[str, ...]:
"""Get the required scopes for a function.
Parameters
----------
func : Callable[..., Any]
The function to inspect.
Returns
-------
scopes : Tuple[:class:`str`, ...]
A tuple of scopes required for a call to succeed.
"""
if not hasattr(func, "__requires_spotify_scopes__"):
raise AttributeError("Scope metadata has not been set for this object!")
return func.__requires_spotify_scopes__ # type: ignore
class OAuth2:
"""Helper object for Spotify OAuth2 operations.
At a very basic level you can you oauth2 only for authentication.
>>> oauth2 = OAuth2(client, 'some://redirect/uri')
>>> print(oauth2.url)
Working with scopes:
>>> oauth2 = OAuth2(client, 'some://redirect/uri', scopes=['user-read-currently-playing'])
>>> oauth2.set_scopes(user_read_playback_state=True)
Parameters
----------
client_id : :class:`str`
The client id provided by spotify for the app.
redirect_uri : :class:`str`
The URI Spotify should redirect to.
scopes : Optional[Iterable[:class:`str`], Dict[:class:`str`, :class:`bool`]]
The scopes to be requested.
state : Optional[:class:`str`]
The state used to secure sessions.
Attributes
----------
attrs : Dict[:class:`str`, :class:`str`]
The attributes used when constructing url parameters
parameters : :class:`str`
The URL parameters used
url : :class:`str`
The URL for OAuth2
"""
_BASE = "https://accounts.spotify.com/authorize/?response_type=code&{parameters}"
def __init__(
self,
client_id: str,
redirect_uri: str,
*,
scopes: Optional[Union[Iterable[str], Dict[str, bool]]] = None,
state: str = None,
):
self.client_id = client_id
self.redirect_uri = redirect_uri
self.state = state
self.__scopes: Set[str] = set()
if scopes is not None:
if not isinstance(scopes, dict) and hasattr(scopes, "__iter__"):
scopes = {scope: True for scope in scopes}
if isinstance(scopes, dict):
self.set_scopes(**scopes)
else:
raise TypeError(
f"scopes must be an iterable of strings or a dict of string to bools"
)
def __repr__(self):
return f"<spotfy.OAuth2: client_id={self.client_id!r}, scope={self.scopes!r}>"
def __str__(self):
return self.url
# Alternate constructors
@classmethod
def from_client(cls, client, *args, **kwargs):
"""Construct a OAuth2 object from a `spotify.Client`."""
return cls(client.http.client_id, *args, **kwargs)
@staticmethod
def url_only(**kwargs) -> str:
"""Construct a OAuth2 URL instead of an OAuth2 object."""
oauth = OAuth2(**kwargs)
return oauth.url
# Properties
@property
def scopes(self):
""":class:`frozenset` - A frozenset of the current scopes"""
return frozenset(self.__scopes)
@property
def attributes(self):
"""Attributes used when constructing url parameters."""
data = {"client_id": self.client_id, "redirect_uri": quote(self.redirect_uri)}
if self.scopes:
data["scope"] = quote(" ".join(self.scopes))
if self.state is not None:
data["state"] = self.state
return data
attrs = attributes
@property
def parameters(self) -> str:
""":class:`str` - The formatted url parameters that are used."""
return "&".join("{0}={1}".format(*item) for item in self.attributes.items())
@property
def url(self) -> str:
""":class:`str` - The formatted oauth url used for authorization."""
return self._BASE.format(parameters=self.parameters)
# Public api
def set_scopes(self, **scopes: Dict[str, bool]) -> None:
r"""Modify the scopes for the current oauth2 object.
Add or remove certain scopes from this oauth2 instance.
Since hypens are not allowed, replace the _ with -.
>>> oauth2.set_scopes(user_read_playback_state=True)
Parameters
----------
\*\*scopes: Dict[:class:`str`, :class:`bool`]
The scopes to enable or disable.
"""
for scope_name, state in scopes.items():
scope_name = scope_name.replace("_","-")
if state:
self.__scopes.add(scope_name)
else:
self.__scopes.remove(scope_name)

16
pomice/spotify/playlist.py Normal file
View File

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

26
pomice/spotify/sync/__init__.py
View File

@ -1,26 +0,0 @@
# pylint: skip-file
from spotify import *
from spotify import __all__, _types, Client
from spotify.utils import clean as _clean_namespace
from . import models
from .models import Client, Synchronous as _Sync
with _clean_namespace(locals(), "name", "base", "Mock"):
for name, base in _types.items():
class Mock(base, metaclass=_Sync): # type: ignore
__slots__ = {"__client_thread__"}
Mock.__name__ = base.__name__
Mock.__qualname__ = base.__qualname__
Mock.__doc__ = base.__doc__
locals()[name] = Mock
setattr(models, name, Mock)
Client._default_http_client = locals()[
"HTTPClient"
] # pylint: disable=protected-access

83
pomice/spotify/sync/models.py
View File

@ -1,83 +0,0 @@
from functools import wraps
from inspect import getmembers, iscoroutinefunction
from typing import Type, Callable, TYPE_CHECKING
from .. import Client as _Client
from .thread import EventLoopThread
if TYPE_CHECKING:
import spotify
def _infer_initializer(base: Type, name: str) -> Callable[..., None]:
"""Infer the __init__ to use for a given :class:`typing.Type` base and :class:`str` name."""
if name in {"HTTPClient", "HTTPUserClient"}:
def initializer(self: "spotify.HTTPClient", *args, **kwargs) -> None:
base.__init__(self, *args, **kwargs)
self.__client_thread__ = kwargs["loop"].__spotify_thread__ # type: ignore
else:
assert name != "Client"
def initializer(self: "spotify.SpotifyBase", client: _Client, *args, **kwargs) -> None: # type: ignore
base.__init__(self, client, *args, **kwargs)
self.__client_thread__ = client.__client_thread__ # type: ignore
return initializer
def _normalize_coroutine_function(corofunc):
if isinstance(corofunc, classmethod):
@classmethod
@wraps(corofunc)
def wrapped(cls, client, *args, **kwargs):
assert isinstance(client, _Client)
client.__client_thread__.run_coroutine_threadsafe(
corofunc(cls, client, *args, **kwargs)
)
else:
@wraps(corofunc)
def wrapped(self, *args, **kwargs):
return self.__client_thread__.run_coroutine_threadsafe(
corofunc(self, *args, **kwargs)
)
return wrapped
class Synchronous(type):
"""Metaclass used for overloading coroutine functions on models."""
def __new__(cls, name, bases, dct):
klass = super().__new__(cls, name, bases, dct)
base = bases[0]
name = base.__name__
if name != "Client":
# Models and the HTTP classes get their __init__ overloaded.
initializer = _infer_initializer(base, name)
setattr(klass, "__init__", initializer)
for ident, obj in getmembers(base):
if not iscoroutinefunction(obj):
continue
setattr(klass, ident, _normalize_coroutine_function(obj))
return klass # type: ignore
class Client(_Client, metaclass=Synchronous):
def __init__(self, *args, **kwargs):
thread = EventLoopThread()
thread.start()
kwargs["loop"] = thread.loop # pylint: disable=protected-access
super().__init__(*args, **kwargs)
self.__thread = self.__client_thread__ = thread

48
pomice/spotify/sync/thread.py
View File

@ -1,48 +0,0 @@
from asyncio import new_event_loop, run_coroutine_threadsafe, set_event_loop
from threading import Thread, RLock, get_ident
from typing import Any, Coroutine
class EventLoopThread(Thread):
"""A surrogate thread that spins an asyncio event loop."""
def __init__(self):
super().__init__(daemon=True)
self.__lock = RLock()
self.__loop = loop = new_event_loop()
loop.__spotify_thread__ = self
# Properties
@property
def loop(self):
return self.__loop
# Overloads
def run(self):
set_event_loop(self.__loop)
self.__loop.run_forever()
# Public API
def run_coroutine_threadsafe(self, coro: Coroutine) -> Any:
"""Like :func:`asyncio.run_coroutine_threadsafe` but for this specific thread."""
# If the current thread is the same
# as the event loop Thread.
#
# then we're in the process of making
# nested calls to await other coroutines
# and should pass back the coroutine as it should be.
if get_ident() == self.ident:
return coro
# Double lock because I haven't looked
# into whether this deadlocks under whatever
# conditions, Best to play it safe.
with self.__lock:
future = run_coroutine_threadsafe(coro, self.__loop)
return future.result()

12
pomice/spotify/track.py Normal file
View File

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

60
pomice/spotify/utils.py
View File

@ -1,60 +0,0 @@
from re import compile as re_compile
from functools import lru_cache
from contextlib import contextmanager
from typing import Iterable, Hashable, TypeVar, Dict, Tuple
__all__ = ("clean", "filter_items", "to_id")
_URI_RE = re_compile(r"^.*:([a-zA-Z0-9]+)$")
_OPEN_RE = re_compile(r"http[s]?:\/\/open\.spotify\.com\/(.*)\/(.*)")
@contextmanager
def clean(mapping: dict, *keys: Iterable[Hashable]):
"""A helper context manager that defers mutating a mapping."""
yield
for key in keys:
mapping.pop(key)
K = TypeVar("K") # pylint: disable=invalid-name
V = TypeVar("V") # pylint: disable=invalid-name
@lru_cache(maxsize=1024)
def _cached_filter_items(data: Tuple[Tuple[K, V], ...]) -> Dict[K, V]:
data_ = {}
for key, value in data:
if value is not None:
data_[key] = value
return data_
def filter_items(data: Dict[K, V]) -> Dict[K, V]:
"""Filter the items of a dict where the value is not None."""
return _cached_filter_items((*data.items(),))
def to_id(value: str) -> str:
"""Get a spotify ID from a URI or open.spotify URL.
Paramters
---------
value : :class:`str`
The value to operate on.
Returns
-------
id : :class:`str`
The Spotify ID from the value.
"""
value = value.strip()
match = _URI_RE.match(value)
if match is None:
match = _OPEN_RE.match(value)
if match is None:
return value
return match.group(2)
return match.group(1)

2
setup.py
View File

@ -6,7 +6,7 @@ with open("README.md") as f:
setuptools.setup(
name="pomice",
author="cloudwithax",
version="1.0.6.2",
version="1.0.7",
url="https://github.com/cloudwithax/pomice",
packages=setuptools.find_packages(),
license="GPL",