10155059950

Committed 30 Jul 2024 02:22AM UTC coverage: 90.975% (-0.1%) from 91.083%

Build # 10155059950

Build Type

Pull #853

github

Committed by

web-flow

Commit Message

Merge 63b26c840 into 6384c05c8

Pull Request Pull Request #853: Upgrade to Django 4.2 (Big Django Update)

Run Details

99 of 115 new or added lines in 42 files covered. (86.09%)

38 existing lines in 8 files now uncovered.

15000 of 16488 relevant lines covered (90.98%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.27

/spotify/api.py

import json
from django.conf import settings
from django.utils import timezone
from cryptography.fernet import Fernet
from spotipy.cache_handler import CacheHandler
from spotipy import SpotifyOAuth, Spotify, SpotifyException, SpotifyOauthError
from . import models


scopes = ['user-read-currently-playing', 'user-read-playback-state', 'user-modify-playback-state']


class DjangoCacheHandler(CacheHandler):
    """ Custom Spotipy cache handler. Saves authentication information to the database. """

    def __init__(self, user, account_id=None):
        self.user = user
        self.account = account_id

    def get_cached_token(self):
        account = models.SpotifyUser.objects.filter(pk=self.account).first()
        if not self.user:
            raise SpotifySessionError("Account owner must be specified to retrieve token")

        if not account:
            account = models.SpotifyUser.objects.create(user=self.user)
            self.account = account.pk

        if self.user != account.user:
            raise SpotifySessionError("Invalid account ID for user %s" % self.user.name)

        if not account.token_info:
            return None

        try:
            cipher_suite = Fernet(settings.CRYPTO_KEY)
        except ValueError:
            raise SpotifySessionError("Cryptographic key is either missing or invalid")
        return json.loads(cipher_suite.decrypt(account.token_info.encode('utf-8')))

    def save_token_to_cache(self, token_info):
        account = models.SpotifyUser.objects.filter(pk=self.account).first()
        if not self.user:
            raise SpotifySessionError("Account owner must be specified to save token")

        if not account:
            account = models.SpotifyUser.objects.create(user=self.user)
            self.account = account.pk

        try:
            cipher_suite = Fernet(settings.CRYPTO_KEY)
        except ValueError:
            raise SpotifySessionError("Cryptographic key is either missing or invalid")
        encrypted_token_info = cipher_suite.encrypt(json.dumps(token_info).encode('utf-8')).decode('utf-8')

        account.token_info = encrypted_token_info
        account.save()


class SpotifySessionError(Exception):
    def __init__(self, message, error=None, error_description=None, *args, **kwargs):
        self.error = error
        self.error_description = error_description
        self.__dict__.update(kwargs)
        super(SpotifySessionError, self).__init__(message, *args, **kwargs)


def get_spotify_session(account, request_user):
    """
    Call this method to get a Spotify session for a specific Spotify account.

    :param account: SpotifyUser object
    :param request_user: The user currently logged in (User object)
    :return: A Spotify object (manages the session and API calls)
    """

    if account.personal and request_user != account.user:
        raise SpotifySessionError("User is not permitted to access this account")

    cache_handler = DjangoCacheHandler(account.user, account.pk)
    state = "u%da%d" % (account.user.pk, account.pk)

    try:
        auth_manager = SpotifyOAuth(client_id=settings.SPOTIFY_CLIENT_ID, client_secret=settings.SPOTIFY_CLIENT_SECRET,
                                    redirect_uri=settings.SPOTIFY_REDIRECT_URI, state=state, scope=' '.join(scopes),
                                    cache_handler=cache_handler)
    except SpotifyOauthError as error:
        raise SpotifyException(403, "SpotifyOAuth", str(error))

    return Spotify(auth_manager=auth_manager)


def get_track(session, identifier):
    """
    Retrieves the metadata for a given item in Spotify

    :param session: A Session object (used for authenticating the request)
    :param identifier: The unique Spotify ID for the item
    :return: Item metadata, if available (Dictionary)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.track(identifier)
    except SpotifyException:
        return None


def add_to_queue(song_request, request_user):
    """
    Attempt to add a track to the Spotify queue

    :param song_request: Song request object
    :param request_user: The user currently logged in (User object)
    :return: None on success; SpotifyException otherwise
    """

    try:
        api = get_spotify_session(song_request.session.user, request_user)
        api.add_to_queue(song_request.identifier)
        song_request.queued = timezone.now()
        song_request.save()
    except SpotifyException as error:
        return error
    return None


def queue_estimate(session, ms=False):
    """
    Attempt to determine wait time for new song requests

    :param session: The corresponding Session object
    :param ms: If True, return the estimate in milliseconds
    :return: Time remaining, if available (Int)
    """

    currently_playing = get_currently_playing(session)
    if not currently_playing:
        return None
    progress = currently_playing['progress_ms']
    item = currently_playing['item']['id']
    current_song = models.SongRequest.objects.filter(identifier=item, session=session).first()
    time_remaining = 0
    if current_song and current_song.queued:
        time_remaining += current_song.duration - progress
        upcoming = models.SongRequest.objects.filter(queued__gt=current_song.queued, session=session)
        for request in upcoming:
            time_remaining += request.duration
        if not ms:
            time_remaining = int(time_remaining / 60000)
        return time_remaining
    return None


def get_currently_playing(session):
    """
    Attempts to get details about what is currently playing in Spotify

    :param session: The corresponding Session object
    :return: Item metadata, if available (Dictionary)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.currently_playing()
    except SpotifyException:
        return None


def get_playback_state(session):
    """
    Retrieves information about the session's current playback state

    :param session: The corresponding Session object
    :return: Playback state information (Dictionary)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.current_playback()
    except SpotifyException:
        return None


def get_available_devices(account):
    """
    Retrieves information about a Spotify user's available devices

    :param account: The corresponding SpotifyUser object
    :return: Device info (List of Dictionaries)
    """

    try:
        api = get_spotify_session(account, account.user)
        return api.devices()
    except SpotifyException:
        return None


def play(session, device=None):
    """
    Starts or resumes playback for a given session.

    :param session: The corresponding Session object
    :param device: The unique id of the device to start playback on
    :return: None (if successful); Error message otherwise (string)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.start_playback(device)
    except SpotifyException as e:
        return e.msg.split('\n')[-1]


def pause(session):
    """
    Pause playback for a given session.

    :param session: The corresponding Session object
    :return: None (if successful); Error message otherwise (string)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.pause_playback()
    except SpotifyException as e:
        return e.msg.split('\n')[-1]


def previous(session):
    """
    Skip to the previous track.

    :param session: The corresponding Session object
    :return: None (if successful); Error message otherwise (string)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.previous_track()
    except SpotifyException as e:
        return e.msg.split('\n')[-1]


def skip(session):
    """
    Skip to the next track.

    :param session: The corresponding Session object
    :return: None (if successful); Error message otherwise (string)
    """

    try:
        api = get_spotify_session(session.user, session.user.user)
        return api.next_track()
    except SpotifyException as e:
        return e.msg.split('\n')[-1]

1	import json	1✔
2	from django.conf import settings	1✔
3	from django.utils import timezone	1✔
4	from cryptography.fernet import Fernet	1✔
5	from spotipy.cache_handler import CacheHandler	1✔
6	from spotipy import SpotifyOAuth, Spotify, SpotifyException, SpotifyOauthError	1✔
7	from . import models	1✔
8
9
10	scopes = ['user-read-currently-playing', 'user-read-playback-state', 'user-modify-playback-state']	1✔
11
12
13	class DjangoCacheHandler(CacheHandler):	1✔
14	""" Custom Spotipy cache handler. Saves authentication information to the database. """
15
16	def __init__(self, user, account_id=None):	1✔
17	self.user = user	1✔
18	self.account = account_id	1✔
19
20	def get_cached_token(self):	1✔
21	account = models.SpotifyUser.objects.filter(pk=self.account).first()	×
22	if not self.user:	×
23	raise SpotifySessionError("Account owner must be specified to retrieve token")	×
24
25	if not account:	×
26	account = models.SpotifyUser.objects.create(user=self.user)	×
27	self.account = account.pk	×
28
29	if self.user != account.user:	×
30	raise SpotifySessionError("Invalid account ID for user %s" % self.user.name)	×
31
32	if not account.token_info:	×
33	return None	×
34
35	try:	×
36	cipher_suite = Fernet(settings.CRYPTO_KEY)	×
37	except ValueError:	×
38	raise SpotifySessionError("Cryptographic key is either missing or invalid")	×
39	return json.loads(cipher_suite.decrypt(account.token_info.encode('utf-8')))	×
40
41	def save_token_to_cache(self, token_info):	1✔
42	account = models.SpotifyUser.objects.filter(pk=self.account).first()	×
43	if not self.user:	×
44	raise SpotifySessionError("Account owner must be specified to save token")	×
45
46	if not account:	×
47	account = models.SpotifyUser.objects.create(user=self.user)	×
48	self.account = account.pk	×
49
50	try:	×
51	cipher_suite = Fernet(settings.CRYPTO_KEY)	×
52	except ValueError:	×
53	raise SpotifySessionError("Cryptographic key is either missing or invalid")	×
54	encrypted_token_info = cipher_suite.encrypt(json.dumps(token_info).encode('utf-8')).decode('utf-8')	×
55
56	account.token_info = encrypted_token_info	×
57	account.save()	×
58
59
60	class SpotifySessionError(Exception):	1✔
61	def __init__(self, message, error=None, error_description=None, args, *kwargs):	1✔
62	self.error = error	×
63	self.error_description = error_description	×
64	self.__dict__.update(kwargs)	×
65	super(SpotifySessionError, self).__init__(message, args, *kwargs)	×
66
67
68	def get_spotify_session(account, request_user):	1✔
69	"""
70	Call this method to get a Spotify session for a specific Spotify account.
71
72	:param account: SpotifyUser object
73	:param request_user: The user currently logged in (User object)
74	:return: A Spotify object (manages the session and API calls)
75	"""
76
77	if account.personal and request_user != account.user:	1✔
78	raise SpotifySessionError("User is not permitted to access this account")	×
79
80	cache_handler = DjangoCacheHandler(account.user, account.pk)	1✔
81	state = "u%da%d" % (account.user.pk, account.pk)	1✔
82
83	try:	1✔
84	auth_manager = SpotifyOAuth(client_id=settings.SPOTIFY_CLIENT_ID, client_secret=settings.SPOTIFY_CLIENT_SECRET,	1✔
85	redirect_uri=settings.SPOTIFY_REDIRECT_URI, state=state, scope=' '.join(scopes),
86	cache_handler=cache_handler)
87	except SpotifyOauthError as error:	1✔
88	raise SpotifyException(403, "SpotifyOAuth", str(error))	1✔
89
90	return Spotify(auth_manager=auth_manager)	×
91
92
93	def get_track(session, identifier):	1✔
94	"""
95	Retrieves the metadata for a given item in Spotify
96
97	:param session: A Session object (used for authenticating the request)
98	:param identifier: The unique Spotify ID for the item
99	:return: Item metadata, if available (Dictionary)
100	"""
101
UNCOV 102	try:	×
UNCOV 103	api = get_spotify_session(session.user, session.user.user)	×
104	return api.track(identifier)	×
UNCOV 105	except SpotifyException:	×
UNCOV 106	return None	×
107
108
109	def add_to_queue(song_request, request_user):	1✔
110	"""
111	Attempt to add a track to the Spotify queue
112
113	:param song_request: Song request object
114	:param request_user: The user currently logged in (User object)
115	:return: None on success; SpotifyException otherwise
116	"""
117
118	try:	1✔
119	api = get_spotify_session(song_request.session.user, request_user)	1✔
120	api.add_to_queue(song_request.identifier)	×
121	song_request.queued = timezone.now()	×
122	song_request.save()	×
123	except SpotifyException as error:	1✔
124	return error	1✔
125	return None	×
126
127
128	def queue_estimate(session, ms=False):	1✔
129	"""
130	Attempt to determine wait time for new song requests
131
132	:param session: The corresponding Session object
133	:param ms: If True, return the estimate in milliseconds
134	:return: Time remaining, if available (Int)
135	"""
136
137	currently_playing = get_currently_playing(session)	1✔
138	if not currently_playing:	1✔
139	return None	1✔
140	progress = currently_playing['progress_ms']	×
141	item = currently_playing['item']['id']	×
142	current_song = models.SongRequest.objects.filter(identifier=item, session=session).first()	×
143	time_remaining = 0	×
144	if current_song and current_song.queued:	×
145	time_remaining += current_song.duration - progress	×
146	upcoming = models.SongRequest.objects.filter(queued__gt=current_song.queued, session=session)	×
147	for request in upcoming:	×
148	time_remaining += request.duration	×
149	if not ms:	×
150	time_remaining = int(time_remaining / 60000)	×
151	return time_remaining	×
152	return None	×
153
154
155	def get_currently_playing(session):	1✔
156	"""
157	Attempts to get details about what is currently playing in Spotify
158
159	:param session: The corresponding Session object
160	:return: Item metadata, if available (Dictionary)
161	"""
162
163	try:	1✔
164	api = get_spotify_session(session.user, session.user.user)	1✔
165	return api.currently_playing()	×
166	except SpotifyException:	1✔
167	return None	1✔
168
169
170	def get_playback_state(session):	1✔
171	"""
172	Retrieves information about the session's current playback state
173
174	:param session: The corresponding Session object
175	:return: Playback state information (Dictionary)
176	"""
177
178	try:	1✔
179	api = get_spotify_session(session.user, session.user.user)	1✔
180	return api.current_playback()	×
181	except SpotifyException:	1✔
182	return None	1✔
183
184
185	def get_available_devices(account):	1✔
186	"""
187	Retrieves information about a Spotify user's available devices
188
189	:param account: The corresponding SpotifyUser object
190	:return: Device info (List of Dictionaries)
191	"""
192
193	try:	1✔
194	api = get_spotify_session(account, account.user)	1✔
195	return api.devices()	×
196	except SpotifyException:	1✔
197	return None	1✔
198
199
200	def play(session, device=None):	1✔
201	"""
202	Starts or resumes playback for a given session.
203
204	:param session: The corresponding Session object
205	:param device: The unique id of the device to start playback on
206	:return: None (if successful); Error message otherwise (string)
207	"""
208
209	try:	1✔
210	api = get_spotify_session(session.user, session.user.user)	1✔
211	return api.start_playback(device)	×
212	except SpotifyException as e:	1✔
213	return e.msg.split('\n')[-1]	1✔
214
215
216	def pause(session):	1✔
217	"""
218	Pause playback for a given session.
219
220	:param session: The corresponding Session object
221	:return: None (if successful); Error message otherwise (string)
222	"""
223
224	try:	1✔
225	api = get_spotify_session(session.user, session.user.user)	1✔
226	return api.pause_playback()	×
227	except SpotifyException as e:	1✔
228	return e.msg.split('\n')[-1]	1✔
229
230
231	def previous(session):	1✔
232	"""
233	Skip to the previous track.
234
235	:param session: The corresponding Session object
236	:return: None (if successful); Error message otherwise (string)
237	"""
238
239	try:	1✔
240	api = get_spotify_session(session.user, session.user.user)	1✔
241	return api.previous_track()	×
242	except SpotifyException as e:	1✔
243	return e.msg.split('\n')[-1]	1✔
244
245
246	def skip(session):	1✔
247	"""
248	Skip to the next track.
249
250	:param session: The corresponding Session object
251	:return: None (if successful); Error message otherwise (string)
252	"""
253
254	try:	1✔
255	api = get_spotify_session(session.user, session.user.user)	1✔
256	return api.next_track()	×
257	except SpotifyException as e:	1✔
258	return e.msg.split('\n')[-1]	1✔

WPI-LNL / lnldb / 10155059950

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous