Source code for compiam.dunya.conn

# Copyright 2013, 2014 Music Technology Group - Universitat Pompeu Fabra
#
# This file is part of Dunya and has been ported from pycompmusic
# (https://github.com/MTG/pycompmusic), the official Python API
#
# Dunya is free software: you can redistribute it and/or modify it under the
# terms of the GNU Affero General Public License as published by the Free Software
# Foundation (FSF), either version 3 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
# PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# this program.  If not, see http://www.gnu.org/licenses/

import logging
import urllib.parse as urllibparse

import requests
import requests.adapters

logger = logging.getLogger("dunya")

from compiam.exceptions import HTTPError, ConnectionError

HOSTNAME = "https://dunya.compmusic.upf.edu"
TOKEN = None
session = requests.Session()
session.mount("http://", requests.adapters.HTTPAdapter(max_retries=5))
session.mount("https://", requests.adapters.HTTPAdapter(max_retries=5))


[docs]def set_hostname(hostname): """Change the hostname of the dunya API endpoint. :param hostname: The new dunya hostname to set. If you want to access over http or a different port, include them in the hostname, e.g. `http://localhost:8000`. :returns: None. """ global HOSTNAME HOSTNAME = hostname
[docs]def set_token(token): """Get an access token. You must call this before you can make. :param token: your access token. :returns: None. """ global TOKEN TOKEN = token
def _dunya_url_query(url, extra_headers=None): """Use requests to query from a given url in Dunya.""" logger.debug("query to '%s'" % url) if not TOKEN: raise ConnectionError("You need to authenticate with `set_token`") headers = {"Authorization": "Token %s" % TOKEN} if extra_headers: headers.update(extra_headers) g = session.get(url, headers=headers) try: g.raise_for_status() except requests.exceptions.HTTPError as e: raise HTTPError(e) return g def _make_url(path, **kwargs): """Create URL from path.""" if "://" in HOSTNAME: protocol, hostname = HOSTNAME.split("://") else: protocol = "http" hostname = HOSTNAME if not kwargs: kwargs = {} for key, value in kwargs.items(): if isinstance(value, str): kwargs[key] = value.encode("utf8") url = urllibparse.urlunparse( (protocol, hostname, "%s" % path, "", urllibparse.urlencode(kwargs), "") ) return url def _dunya_query_json(path, **kwargs): """Make a query to dunya and expect the results to be JSON.""" g = _dunya_url_query(_make_url(path, **kwargs)) return g.json() if g else None def _dunya_query_file(path, **kwargs): """Make a query to dunya and return the raw result.""" g = _dunya_url_query(_make_url(path, **kwargs)) if g: cl = g.headers.get("content-length") content = g.content if cl and int(cl) != len(content): logger.warning( "Indicated content length is not the same as returned content. Some data may be missing" ) return content else: return def _file_for_document(recording_id, thetype, subtype=None, part=None, version=None): """Get the most recent derived file given a filetype. :param recording_id: Musicbrainz recording ID. :param thetype: the computed filetype. :param subtype: a subtype if the module has one. :param part: the file part if the module has one. :param version: a specific version, otherwise the most recent one will be used. :returns: The contents of the most recent version of the derived file. """ path = "document/by-id/%s/%s" % (recording_id, thetype) args = {} if subtype: args["subtype"] = subtype if part: args["part"] = part if version: args["v"] = version return _dunya_query_file(path, **args)
[docs]def get_mp3(recording_id): """Get a mp3 from a specific mbid. :param recording_id: Musicbrainz recording ID. """ return _file_for_document(recording_id, "mp3")