File: _backendrest.py

package info (click to toggle)
python-bugzilla 3.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye, sid
  • size: 1,072 kB
  • sloc: python: 5,760; makefile: 9
file content (193 lines) | stat: -rw-r--r-- 6,735 bytes parent folder | download
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)