...
 
Commits (3)
......@@ -15,6 +15,8 @@ Features
--------
* Searching for tracks, albums and artists available in your Funkwhale instance
* Browse all artists and albums
* Browse your favorites
* Simple configuration
Installation
......@@ -47,15 +49,19 @@ To enable the extension, add the following to your ``mopidy.conf`` file::
username = demo
# Password to use when authenticating (leave empty fo anonymous access)
password = demo
# duration of cache entries before they are removed, in seconds
# 0 to cache forever, empty to disable cache
cache_duration = 600
Of course, replace the demo values with your actual info (but you can
try using the demo server).
After that, reload your mopidy daemon, and you should be good!
Todo
----
- Browse Funkwhale library and playlists
- Browse use library and playlists
.. _Mopidy: https://www.mopidy.com/
......
......@@ -9,39 +9,41 @@ import os
__author__ = """Eliot Berriot"""
__email__ = 'contact+funkwhale@eliotberriot.com'
__version__ = '0.1.0'
__email__ = "contact+funkwhale@eliotberriot.com"
__version__ = "0.1.0"
logger = logging.getLogger(__name__)
class Extension(mopidy.ext.Extension):
dist_name = 'Mopidy-Funkwhale'
ext_name = 'funkwhale'
dist_name = "Mopidy-Funkwhale"
ext_name = "funkwhale"
version = __version__
def get_default_config(self):
conf_file = os.path.join(os.path.dirname(__file__), 'ext.conf')
conf_file = os.path.join(os.path.dirname(__file__), "ext.conf")
return mopidy.config.read(conf_file)
def get_config_schema(self):
schema = super(Extension, self).get_config_schema()
schema['url'] = mopidy.config.String()
schema['username'] = mopidy.config.String(optional=True)
schema['password'] = mopidy.config.Secret(optional=True)
schema["url"] = mopidy.config.String()
schema["username"] = mopidy.config.String(optional=True)
schema["password"] = mopidy.config.Secret(optional=True)
schema["cache_duration"] = mopidy.config.Integer(optional=True)
return schema
def validate_config(self, config):
if not config.getboolean('funkwhale', 'enabled'):
if not config.getboolean("funkwhale", "enabled"):
return
username = config.getstring('funkwhale', 'username')
password = config.getstring('funkwhale', 'password')
username = config.getstring("funkwhale", "username")
password = config.getstring("funkwhale", "password")
if any([username, password]) and not all([username, password]):
raise mopidy.ext.ExtensionError(
'You need to provide username and password to authenticate with the funkwhale backend'
"You need to provide username and password to authenticate with the funkwhale backend"
)
def setup(self, registry):
from . import actor
registry.add('backend', actor.FunkwhaleBackend)
registry.add("backend", actor.FunkwhaleBackend)
......@@ -14,34 +14,37 @@ logger = logging.getLogger(__name__)
class FunkwhaleBackend(pykka.ThreadingActor, backend.Backend):
def __init__(self, config, audio):
super(FunkwhaleBackend, self).__init__()
self.config = config
self.remote = client.FunkwhaleClient(config)
self.client = client.APIClient(config)
self.library = library.FunkwhaleLibraryProvider(backend=self)
self.playback = FunkwhalePlaybackProvider(audio=audio, backend=self)
self.uri_schemes = ['funkwhale', 'fw']
self.uri_schemes = ["funkwhale", "fw"]
def on_start(self):
username = self.remote.username
if username is not None:
logger.info('Logged in to Funkwhale as "%s" on "%s"', username, self.config['funkwhale']['url'])
if self.client.username is not None:
self.client.login()
logger.info(
'Logged in to Funkwhale as "%s" on "%s"',
self.client.username,
self.config["funkwhale"]["url"],
)
else:
logger.info('Using "%s" anonymously', self.config['funkwhale']['url'])
logger.info('Using "%s" anonymously', self.config["funkwhale"]["url"])
class FunkwhalePlaybackProvider(backend.PlaybackProvider):
def translate_uri(self, uri):
_, id = library.parse_uri(uri)
track = self.backend.remote.http_client.get_track(id)
track = self.backend.client.get_track(id)
if track is None:
return None
url = track['listen_url']
if url.startswith('/'):
url = self.backend.config['funkwhale']['url'] + url
if self.backend.remote.token:
url += '?jwt=' + self.backend.remote.token
url = track["listen_url"]
if url.startswith("/"):
url = self.backend.config["funkwhale"]["url"] + url
if self.backend.client.token:
url += "?jwt=" + self.backend.client.token
return url
from __future__ import unicode_literals
import logging
import requests
from mopidy import httpclient, exceptions
from . import Extension, __version__
logger = logging.getLogger(__name__)
class SessionWithUrlBase(requests.Session):
# In Python 3 you could place `url_base` after `*args`, but not in Python 2.
......@@ -17,74 +20,103 @@ class SessionWithUrlBase(requests.Session):
# 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
if url.startswith("http://") or url.startswith("https://"):
modified_url = url
else:
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/'
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})
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})
if not username:
return
response = session.post("token/", {"username": username, "password": password})
try:
response.raise_for_status()
except requests.exceptions.HTTPError:
raise exceptions.BackendError('Authentication failed for user %s' % (username,))
token = response.json()['token']
session.headers.update({'Authorization': 'JWT %s' % (token,)})
raise exceptions.BackendError("Authentication failed for user %s" % (username,))
token = response.json()["token"]
session.headers.update({"Authorization": "JWT %s" % (token,)})
return token
class APIClient(object):
def __init__(self, session):
self.session = session
def __init__(self, config):
self.config = config
self.session = get_requests_session(
config["funkwhale"]["url"],
proxy_config=config["proxy"],
user_agent="%s/%s" % (Extension.dist_name, __version__),
)
self.username = self.config["funkwhale"]["username"]
self.token = None
def login(self):
self.username = self.config["funkwhale"]["username"]
if self.username:
self.token = login(
self.session,
self.config["funkwhale"]["username"],
self.config["funkwhale"]["password"],
)
else:
self.token = None
def search(self, query):
response = self.session.get('search', params={'query': 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 = 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 = self.session.get("tracks/", params=filters)
response.raise_for_status()
return response.json()
def list_artists(self, filters):
response = self.session.get("artists/", params=filters)
response.raise_for_status()
return response.json()
class FunkwhaleClient(object):
def list_albums(self, filters):
response = self.session.get("albums/", params=filters)
response.raise_for_status()
return response.json()
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']
self.token = None
if config['funkwhale']['username']:
self.token = login(
self.http_client.session,
config['funkwhale']['username'],
config['funkwhale']['password'])
def load_all(self, first_page, max=0):
for i in first_page["results"]:
yield i
next_page = first_page.get("next")
counter = 0
while next_page:
logger.info("Fetching next page of result at url: %s", next_page)
response = self.session.get(next_page)
response.raise_for_status()
payload = response.json()
for i in payload["results"]:
yield i
counter += 1
next_page = payload.get("next")
if max and counter >= max:
next_page = None
......@@ -6,3 +6,6 @@ url = https://demo.funkwhale.audio
username = demo
# Password to use when authenticating (leave empty fo anonymous access)
password = demo
# duration of cache entries before they are removed, in seconds
# 0 to cache forever, empty to disable cache
cache_duration = 600
This diff is collapsed.
......@@ -35,6 +35,7 @@ test =
pytest-cov
requests-mock
pytest-mock
factory_boy
dev =
ipython
......
import pytest
import mopidy_funkwhale.actor
import mopidy_funkwhale.client
import mopidy_funkwhale.library
FUNKWHALE_URL = "https://test.funkwhale"
@pytest.fixture()
def config():
return {
"funkwhale": {
"url": FUNKWHALE_URL,
"username": "user",
"password": "passw0rd",
"cache_duration": 600,
},
"proxy": {},
}
@pytest.fixture
def backend(config):
return mopidy_funkwhale.actor.FunkwhaleBackend(config=config, audio=None)
@pytest.fixture()
def session(backend):
return mopidy_funkwhale.client.get_requests_session(
FUNKWHALE_URL, {}, "test/something"
)
@pytest.fixture()
def client(backend, session):
return backend.client
@pytest.fixture
def library(backend):
return backend.library
import random
from mopidy import models
import factory
class ArtistJSONFactory(factory.Factory):
id = factory.Sequence(int)
mbid = factory.Faker("uuid4")
name = factory.Faker("name")
class Meta:
model = dict
class CoverJSONFactory(factory.Factory):
original = factory.Faker("url")
class Meta:
model = dict
class AlbumJSONFactory(factory.Factory):
id = factory.Sequence(int)
mbid = factory.Faker("uuid4")
title = factory.Faker("name")
tracks = factory.Iterator([range(i) for i in range(1, 30)])
artist = factory.SubFactory(ArtistJSONFactory)
release_date = factory.Faker("date")
cover = factory.SubFactory(CoverJSONFactory)
class Meta:
model = dict
class TrackJSONFactory(factory.Factory):
id = factory.Sequence(int)
mbid = factory.Faker("uuid4")
title = factory.Faker("name")
position = factory.Faker("pyint")
duration = factory.Faker("pyint")
creation_date = factory.Faker("date")
bitrate = factory.Iterator([i * 1000 for i in (128, 256, 360)])
artist = factory.SubFactory(ArtistJSONFactory)
album = factory.SubFactory(AlbumJSONFactory)
class Meta:
model = dict
class ArtistFactory(factory.Factory):
class Meta:
model = models.Artist
import pytest
import mopidy_funkwhale.client
FUNKWHALE_URL = 'https://test.funkwhale'
FUNKWHALE_API_URL = FUNKWHALE_URL + '/api/v1/'
def test_client_search(client, requests_mock):
requests_mock.get(
client.session.url_base + "search?query=myquery", json={"hello": "world"}
)
result = client.search("myquery")
assert result == {"hello": "world"}
@pytest.fixture()
def session():
return mopidy_funkwhale.client.get_requests_session(
FUNKWHALE_URL,
{},
'test/something'
)
def test_client_get_track(client, requests_mock):
requests_mock.get(client.session.url_base + "tracks/12/", json={"hello": "world"})
@pytest.fixture()
def client(session):
return mopidy_funkwhale.client.APIClient(session)
result = client.get_track(12)
assert result == {"hello": "world"}
def test_client_search(client, requests_mock):
requests_mock.get(FUNKWHALE_API_URL + 'search?query=myquery', json={'hello': 'world'})
def test_client_list_tracks(client, requests_mock):
requests_mock.get(
client.session.url_base + "tracks/?artist=12", json={"hello": "world"}
)
result = client.search('myquery')
assert result == {'hello': 'world'}
result = client.list_tracks({"artist": 12})
assert result == {"hello": "world"}
def test_client_get_track(client, requests_mock):
requests_mock.get(FUNKWHALE_API_URL + 'tracks/12/', json={'hello': 'world'})
def test_client_list_artists(client, requests_mock):
requests_mock.get(
client.session.url_base + "artists/?playable=true", json={"hello": "world"}
)
result = client.get_track(12)
assert result == {'hello': 'world'}
result = client.list_artists({"playable": "true"})
assert result == {"hello": "world"}
def test_client_list_albums(client, requests_mock):
requests_mock.get(
client.session.url_base + "albums/?playable=true", json={"hello": "world"}
)
result = client.list_albums({"playable": "true"})
assert result == {"hello": "world"}
def test_client_list_tracks(client, requests_mock):
requests_mock.get(FUNKWHALE_API_URL + 'tracks/?artist=12', json={'hello': 'world'})
result = client.list_tracks({'artist': 12})
assert result == {'hello': 'world'}
def test_load_all(client, requests_mock):
page1 = {"results": [1, 2, 3], "next": "https://first.page"}
page2 = {"results": [4, 5, 6], "next": "https://second.page"}
page3 = {"results": [7, 8, 9], "next": None}
requests_mock.get(page1["next"], json=page2)
requests_mock.get(page2["next"], json=page3)
assert (
list(client.load_all(page1))
== page1["results"] + page2["results"] + page3["results"]
)
......@@ -8,17 +8,18 @@ def test_get_default_config():
config = ext.get_default_config()
assert '[funkwhale]' in config
assert 'enabled = true' in config
assert 'url = https://demo.funkwhale.audio' in config
assert 'username = demo' in config
assert 'password = demo' in config
assert "[funkwhale]" in config
assert "enabled = true" in config
assert "url = https://demo.funkwhale.audio" in config
assert "username = demo" in config
assert "password = demo" in config
def test_get_config_schema():
ext = mopidy_funkwhale.Extension()
schema = ext.get_config_schema()
assert 'url' in schema
assert 'username' in schema
assert 'password' in schema
assert "url" in schema
assert "username" in schema
assert "password" in schema
assert "cache_duration" in schema
This diff is collapsed.