FilesExpand file tree

 main

/

_backendrest.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

229 lines (191 loc) · 8.09 KB

 main

/

_backendrest.py

Copy path

Top

File metadata and controls

• Code
• Blame

229 lines (191 loc) · 8.09 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

# 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, BugzillaHTTPError

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_error(self, e):

response = getattr(e, "response", None)

if response is None:

raise e # pragma: no cover

if response.status_code in [400, 401, 404]:

self._handle_error_response(response.text)

raise e

def _handle_error_response(self, text):

try:

result = json.loads(text)

except json.JSONDecodeError:

return

if result.get("error"):

raise BugzillaError(result["message"], code=result["code"])

def _handle_response(self, text):

try:

ret = dict(json.loads(text))

except Exception: # pragma: no cover

log.debug("Failed to parse REST response. Output is:\n%s", text)

raise

if ret.get("error", False): # pragma: no cover

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

authparams = self._bugzillasession.get_auth_params()

if method == "GET":

authparams.update(paramdict or {})

else:

data = json.dumps(paramdict or {})

try:

response = self._bugzillasession.request(

method, fullurl, data=data, params=authparams

)

except BugzillaHTTPError as e:

self._handle_error(e)

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):

bug_list = listify(bug_ids)

alias_list = listify(aliases)

permissive = paramdict.pop("permissive", False)

data = paramdict.copy()

# FYI: The high-level API expects the backends to raise an exception

# when retrieval of a single bug fails (default behavior of the XMLRPC

# API), but the REST API simply returns an empty search result set.

# To ensure compliant behavior, the REST backend needs to use the

# explicit URL to get a single bug.

if not permissive and len(bug_list or []) + len(alias_list or []) == 1:

for id_list in (bug_list, alias_list):

if id_list:

return self._get("/bug/%s" % id_list[0], data)

data["id"] = bug_list

data["alias"] = alias_list

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]) # pragma: no cover

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): # pragma: no cover

raise BugzillaError(

"No REST API available yet for externalbugs_add")

def externalbugs_remove(self, paramdict): # pragma: no cover

raise BugzillaError(

"No REST API available yet for externalbugs_remove")

def externalbugs_update(self, paramdict): # pragma: no cover

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] # pragma: no cover

if "names" in paramdict:

urlid = listify(paramdict["names"])[0]

return self._put("/user/%s" % urlid, paramdict)