-
Notifications
You must be signed in to change notification settings - Fork 10
/
_api.py
82 lines (64 loc) · 2.36 KB
/
_api.py
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
"""Heavily derived from https://github.com/biocore/microsetta-private-api/blob/minimalInterface/microsetta_private_api/example/client_impl.py""" # noqa
import requests
from microsetta_admin.config_manager import SERVER_CONFIG
from flask import redirect, session
from urllib.parse import urljoin
TOKEN_KEY_NAME = 'token'
class BearerAuth(requests.auth.AuthBase):
def __init__(self, token):
self.token = token
def __call__(self, r):
r.headers['Authorization'] = "Bearer " + self.token
return r
class APIRequest:
API_URL = SERVER_CONFIG["private_api_url"]
DEFAULT_PARAMS = {'language_tag': 'en-US'}
CAfile = SERVER_CONFIG["CAfile"]
@classmethod
def build_params(cls, params):
all_params = {}
for key in cls.DEFAULT_PARAMS:
all_params[key] = cls.DEFAULT_PARAMS[key]
all_params.update({} if params is None else params)
return all_params
@staticmethod
def _check_response(response):
output = None
if response.status_code == 401:
# redirect to home page for login
output = redirect("/")
elif response.status_code >= 400:
if response.text:
output = response.text
else:
output = "unknown error"
else:
if response.text:
output = response.json()
return response.status_code, output
@classmethod
def get(cls, path, params=None):
response = requests.get(
urljoin(cls.API_URL, path),
auth=BearerAuth(session[TOKEN_KEY_NAME]),
verify=cls.CAfile,
params=cls.build_params(params))
return cls._check_response(response)
@classmethod
def put(cls, path, params=None, json=None):
response = requests.put(
urljoin(cls.API_URL, path),
auth=BearerAuth(session[TOKEN_KEY_NAME]),
verify=cls.CAfile,
params=cls.build_params(params),
json=json)
return cls._check_response(response)
@classmethod
def post(cls, path, params=None, json=None):
response = requests.post(
urljoin(cls.API_URL, path),
auth=BearerAuth(session[TOKEN_KEY_NAME]),
verify=cls.CAfile,
params=cls.build_params(params),
json=json)
return cls._check_response(response)