-
Notifications
You must be signed in to change notification settings - Fork 114
Expand file tree
/
Copy path_backendrest.py
More file actions
193 lines (161 loc) · 6.58 KB
/
_backendrest.py
File metadata and controls
193 lines (161 loc) · 6.58 KB
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.
import base64
import json
import logging
import os
from ._backendbase import _BackendBase
from .exceptions import BugzillaError
from ._util import listify
log = logging.getLogger(__name__)
def _update_key(indict, updict, key):
if key not in indict:
indict[key] = {}
indict[key].update(updict.get(key, {}))
class _BackendREST(_BackendBase):
"""
Internal interface for direct calls to bugzilla's REST API
"""
def __init__(self, url, bugzillasession):
_BackendBase.__init__(self, url, bugzillasession)
self._bugzillasession.set_rest_defaults()
#########################
# Internal REST helpers #
#########################
def _handle_response(self, text):
try:
ret = dict(json.loads(text))
except Exception:
log.debug("Failed to parse REST response. Output is:\n%s", text)
raise
if ret.get("error", False):
raise BugzillaError(ret["message"], code=ret["code"])
return ret
def _op(self, method, apiurl, paramdict=None):
fullurl = os.path.join(self._url, apiurl.lstrip("/"))
log.debug("Bugzilla REST %s %s params=%s", method, fullurl, paramdict)
data = None
params = None
if method == "GET":
params = paramdict
else:
data = json.dumps(paramdict or {})
response = self._bugzillasession.request(method, fullurl, data=data,
params=params)
return self._handle_response(response.text)
def _get(self, *args, **kwargs):
return self._op("GET", *args, **kwargs)
def _put(self, *args, **kwargs):
return self._op("PUT", *args, **kwargs)
def _post(self, *args, **kwargs):
return self._op("POST", *args, **kwargs)
#######################
# API implementations #
#######################
def get_xmlrpc_proxy(self):
raise BugzillaError("You are using the bugzilla REST API, "
"so raw XMLRPC access is not provided.")
def is_rest(self):
return True
def bugzilla_version(self):
return self._get("/version")
def bug_create(self, paramdict):
return self._post("/bug", paramdict)
def bug_fields(self, paramdict):
return self._get("/field/bug", paramdict)
def bug_get(self, bug_ids, aliases, paramdict):
data = paramdict.copy()
data["id"] = listify(bug_ids)
data["alias"] = listify(aliases)
ret = self._get("/bug", data)
return ret
def bug_attachment_get(self, attachment_ids, paramdict):
# XMLRPC supported mutiple fetch at once, but not REST
ret = {}
for attid in listify(attachment_ids):
out = self._get("/bug/attachment/%s" % attid, paramdict)
_update_key(ret, out, "attachments")
_update_key(ret, out, "bugs")
return ret
def bug_attachment_get_all(self, bug_ids, paramdict):
# XMLRPC supported mutiple fetch at once, but not REST
ret = {}
for bugid in listify(bug_ids):
out = self._get("/bug/%s/attachment" % bugid, paramdict)
_update_key(ret, out, "attachments")
_update_key(ret, out, "bugs")
return ret
def bug_attachment_create(self, bug_ids, data, paramdict):
if data is not None and "data" not in paramdict:
paramdict["data"] = base64.b64encode(data).decode("utf-8")
paramdict["ids"] = listify(bug_ids)
return self._post("/bug/%s/attachment" % paramdict["ids"][0],
paramdict)
def bug_attachment_update(self, attachment_ids, paramdict):
paramdict["ids"] = listify(attachment_ids)
return self._put("/bug/attachment/%s" % paramdict["ids"][0], paramdict)
def bug_comments(self, bug_ids, paramdict):
# XMLRPC supported mutiple fetch at once, but not REST
ret = {}
for bugid in bug_ids:
out = self._get("/bug/%s/comment" % bugid, paramdict)
_update_key(ret, out, "bugs")
return ret
def bug_history(self, bug_ids, paramdict):
# XMLRPC supported mutiple fetch at once, but not REST
ret = {"bugs": []}
for bugid in bug_ids:
out = self._get("/bug/%s/history" % bugid, paramdict)
ret["bugs"].extend(out.get("bugs", []))
return ret
def bug_search(self, paramdict):
return self._get("/bug", paramdict)
def bug_update(self, bug_ids, paramdict):
data = paramdict.copy()
data["ids"] = listify(bug_ids)
return self._put("/bug/%s" % data["ids"][0], data)
def bug_update_tags(self, bug_ids, paramdict):
raise BugzillaError("No REST API available for bug_update_tags")
def component_create(self, paramdict):
return self._post("/component", paramdict)
def component_update(self, paramdict):
if "ids" in paramdict:
apiurl = str(listify(paramdict["ids"])[0])
if "names" in paramdict:
apiurl = ("%(product)s/%(component)s" %
listify(paramdict["names"])[0])
return self._put("/component/%s" % apiurl, paramdict)
def externalbugs_add(self, paramdict):
raise BugzillaError(
"No REST API available yet for externalbugs_add")
def externalbugs_remove(self, paramdict):
raise BugzillaError(
"No REST API available yet for externalbugs_remove")
def externalbugs_update(self, paramdict):
raise BugzillaError(
"No REST API available yet for externalbugs_update")
def group_get(self, paramdict):
return self._get("/group", paramdict)
def product_get(self, paramdict):
return self._get("/product/get", paramdict)
def product_get_accessible(self):
return self._get("/product_accessible")
def product_get_enterable(self):
return self._get("/product_enterable")
def product_get_selectable(self):
return self._get("/product_selectable")
def user_create(self, paramdict):
return self._post("/user", paramdict)
def user_get(self, paramdict):
return self._get("/user", paramdict)
def user_login(self, paramdict):
return self._get("/login", paramdict)
def user_logout(self):
return self._get("/logout")
def user_update(self, paramdict):
urlid = None
if "ids" in paramdict:
urlid = listify(paramdict["ids"])[0]
if "names" in paramdict:
urlid = listify(paramdict["names"])[0]
return self._put("/user/%s" % urlid, paramdict)