python-stackpath/stackpath/stackpath.py at master · koty/python-stackpath · GitHub

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
fromrequests_oauthlibimportOAuth1SessionasOAuth1

# handle python 3.x
try:
importurlparse
exceptImportError:
fromurllibimportparseasurlparse


classStackPath(object):
def__init__(self, alias, key, secret, server="api.stackpath.com", api_version='v1', **kwargs):
self.url="https://%s/%s/%s"% (server, api_version, alias)
self.client=OAuth1(key, client_secret=secret, **kwargs)

def_get_headers(self, json=True):
headers= {"User-Agent": "Python StackPath API Client"}
ifjson:
headers["Content-Type"] ="application/json"
returnheaders

def_get_url(self, end_point):
ifnotend_point.startswith("/"):
return"{0}/{1}".format(self.url, end_point)
else:
returnself.url+end_point

def_parse_json(self, response):
try:
returnresponse.json()
exceptValueErrorase:
raiseself.ServerError(response, str(e))

def_data_request(self, method, end_point, data, **kwargs):
ifdataisNoneand"params"inkwargs:
params=kwargs.pop("params")
iftype(params) isstr:
params=urlparse.parse_qs(params)
data=params
action=getattr(self.client, method)
response=action(self._get_url(end_point), data=data,
headers=self._get_headers(json=True), **kwargs)

ifresponse.status_code>299:
raiseself.ServerError(response)

returnself._parse_json(response)

defget(self, end_point, data=None, **kwargs):
returnself._data_request("get", end_point, data=data, **kwargs)

defpatch(self, end_point, data=None, **kwargs):
returnself._data_request("post", end_point, data=data, **kwargs)

defpost(self, end_point, data=None, **kwargs):
returnself._data_request("post", end_point, data=data, **kwargs)

defput(self, end_point, data=None, **kwargs):
returnself._data_request("put", end_point, data=data, **kwargs)

defdelete(self, end_point, data=None, **kwargs):
returnself._data_request("delete", end_point, data=data, **kwargs)

defpurge(self, site_id, file_or_files=None, **kwargs):
# https://api.stackpath.com/v1/{companyalias}/sites/{site_id}/cache
path="/sites/%s/cache"% (site_id)
iffile_or_filesisnotNone:
returnself.delete(path, data={"files": file_or_files}, **kwargs)
returnself.delete(path, **kwargs)

classServerError(Exception):
def__init__(self, response, message=None):
try:
resp=response.json()
ifmessageisNone:
message="{0}:: {1}".format(resp['error']['type'],
resp['error']['message'])
self.reason=resp['error']['type']
exceptValueError:
ifmessageisNone:
message="{0} {1} from {2}".format(response.status_code,
response.reason,
response.url)

self.reason=response.reason

self.headers=response.headers
self.code=response.status_code
self.body=response._content
self.url=response.url

super(Exception, self).__init__(message)