1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
from datetime import datetime
from ampache import models, exceptions
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):
"""
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.
"""
if error.code == exceptions.ACCESS_ERROR_CODE:
raise exceptions.AccessException(error.message)
elif error.code == exceptions.AUTHENTICATION_ERROR_CODE:
raise exceptions.AuthenticationException(error.message)
elif error.code == exceptions.ACCESS_DENIED_ERROR_CODE:
raise exceptions.AccessDeniedException(error.message)
elif error.code == exceptions.NOT_FOUND_ERROR_CODE:
raise exceptions.NotFoundException(error.message)
elif error.code == exceptions.MISSING_ERROR_CODE:
raise exceptions.MissingMethodException(error.message)
elif error.code == exceptions.DEPRECIATED_ERROR_CODE:
raise exceptions.DepreciatedMethodException(error.message)
elif error.code == exceptions.BAD_REQUEST_ERROR_CODE:
raise exceptions.BadRequestException(error.message)
elif error.code == exceptions.FAILED_ACCESS_ERROR_CODE:
raise exceptions.FailedAccessException(error.message)
def request(self, params, headers):
"""
All in one function to pass JSON requests to the API.
"""
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:
self.raise_by_status(models.Error(int(data['error']['errorCode']),
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.auth,
'filter': song_id,
'version': self._version
}
headers = {
'Content-type': 'application/json'
}
data = self.request(params, headers)
song = models.Song(data['id'], data['title'], data['album']['name'],
data['albumartist']['name'])
return song
def renew_token(self):
if datetime.now(self._auth.expires.tzinfo) > self._auth.expires:
self.authenticate()
def get_token(self):
return self._auth
|
