from http import client from urllib import parse from datetime import datetime import exceptions import models import json import requests def get_utc_offset(time): """ All the datetimes the API yields are in the form yyyy-mm-ddthh:mm:ss-00:00, where the last 5 characters are the UTC offset. Python datetime objects can parse this data, but in the form 0000 (sans the :) so we have to remove it. """ prefix = time[:len(time) - 5] suffix = time[len(time) - 5:].replace(':', '') return prefix + suffix class AmpacheClient: """ The ampache client class. Responsible for making API calls to your ampache instance. """ def __init__(self, host, api_key, version): """ Host: url of your apache instance. api_key: your api key (you can get it under settings->account). It's preferable that you provide an admin's api key. """ self._endpoint = '/server/json.server.php?' self._version = version self._host = host self._api_key = api_key self.authenticate() def raise_by_status(self, error_code): """ The ampache API has a bunch of error codes that can help us diagnose the problem. Depending of the error code, we'll raise an appropiate exception. """ pass def request(self, params, headers): """ All in one function to pass JSON requests to the API. """ self.renew_token() response = requests.get(self._host + self._endpoint, params=params, headers=headers) if not response.ok: response.raise_for_status() data = response.json() if 'error' in data: raise possible_exception(data['error']['errorMessage']) return data def authenticate(self): """ Authenticate with the API, setting the token. """ params = { 'action': 'handshake', 'auth': self._api_key, 'version': self._version } headers = { 'Content-type': 'application/json' } data = self.request(params, headers) self._auth = models.AuthToken(datetime.strptime(get_utc_offset(data['session_expire']), '%Y-%m-%dT%H:%M:%S%z'), data['auth']) def get_song(self, song_id): """ Get the detail of a song given its primary key. """ self.renew_token() params = { 'action': 'song', 'auth': self._auth, 'filter': song_id, 'version': self._version } headers = { 'Content-type': 'application/json' } data = self.request(params, headers) def renew_token(self): if datetime.now(self._auth.expires.tzinfo) > self._auth.expires: self.authenticate() def get_token(self): return self._auth