90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
from __future__ import unicode_literals
|
|
|
|
import requests
|
|
|
|
from mopidy import httpclient
|
|
|
|
from . import Extension, __version__
|
|
|
|
|
|
class SessionWithUrlBase(requests.Session):
|
|
# In Python 3 you could place `url_base` after `*args`, but not in Python 2.
|
|
def __init__(self, url_base=None, *args, **kwargs):
|
|
super(SessionWithUrlBase, self).__init__(*args, **kwargs)
|
|
self.url_base = url_base
|
|
|
|
def request(self, method, url, **kwargs):
|
|
# Next line of code is here for example purposes only.
|
|
# You really shouldn't just use string concatenation here,
|
|
# take a look at urllib.parse.urljoin instead.
|
|
modified_url = self.url_base + url
|
|
|
|
return super(SessionWithUrlBase, self).request(method, modified_url, **kwargs)
|
|
|
|
|
|
def get_requests_session(url, proxy_config, user_agent):
|
|
if not url.endswith('/'):
|
|
url += '/'
|
|
url += 'api/v1/'
|
|
|
|
proxy = httpclient.format_proxy(proxy_config)
|
|
full_user_agent = httpclient.format_user_agent(user_agent)
|
|
|
|
session = SessionWithUrlBase(url_base=url)
|
|
session.proxies.update({'http': proxy, 'https': proxy})
|
|
session.headers.update({'user-agent': full_user_agent})
|
|
|
|
return session
|
|
|
|
|
|
def login(session, username, password):
|
|
response = session.post('token/', {'username': username, 'password': password})
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.exceptions.HTTPError:
|
|
raise BackendError('Authentication failed for user %s' % (username,))
|
|
token = response.json()['token']
|
|
session.headers.update({'Authorization': 'JWT %s' % (token,)})
|
|
|
|
|
|
class APIClient(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
def search(self, query):
|
|
response = self.session.get('search', params={'query': query})
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_track(self, id):
|
|
response = self.session.get('tracks/{}/'.format(id))
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def list_tracks(self, filters):
|
|
response = self.session.get('tracks/', params=filters)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
class FunkwhaleClient(object):
|
|
|
|
def __init__(self, config):
|
|
super(FunkwhaleClient, self).__init__()
|
|
self.page_size = config['funkwhale'].get('page_size', 50)
|
|
self.http_client = APIClient(
|
|
session=get_requests_session(
|
|
config['funkwhale']['url'],
|
|
proxy_config=config['proxy'],
|
|
user_agent='%s/%s' % (
|
|
Extension.dist_name,
|
|
__version__),
|
|
)
|
|
)
|
|
self.username = config['funkwhale']['username']
|
|
if config['funkwhale']['username']:
|
|
login(
|
|
self.http_client.session,
|
|
config['funkwhale']['username'],
|
|
config['funkwhale']['password'])
|