projectal.entities.webhook

 1from projectal import api
 2from projectal.entity import Entity
 3
 4
 5class Webhook(Entity):
 6    """
 7    Implementation of the [Webhook](https://projectal.com/docs/latest/#tag/Webhook) API.
 8    """
 9
10    _path = "webhook"
11    _name = "webhook"
12
13    @classmethod
14    def update(cls, entities):
15        # The webhook API differs from the rest of the system. We need to send some
16        # mandatory fields over even if they haven't changed. Do this by faking
17        # the change history to always include the required fields.
18        if isinstance(entities, dict):
19            e_list = [entities]
20        else:
21            e_list = entities
22
23        for e in e_list:
24            if isinstance(e, Webhook):
25                e._Entity__old.pop("entity", None)
26                e._Entity__old.pop("action", None)
27                e._Entity__old.pop("url", None)
28        return super(Webhook, cls).update(e_list)
29
30    @classmethod
31    def list(cls, start=0, limit=101, ksort="entity", order="asc"):
32        """
33        Get a list of registered webhooks.
34
35        Optionally specify a range for pagination.
36        """
37        url = "/api/webhook/list?start={}&limit={}&ksort={}&order={}".format(
38            start, limit, ksort, order
39        )
40        return api.get(url)
41
42    @classmethod
43    def list_events(cls, **kwargs):
44        """
45        Get a list of webhook events.
46        Use parameter format=False to return eventTime as UTC timestamp.
47        """
48
49        url = "/api/webhookevent/list"
50
51        params = [f"{k}={v}" for k, v in kwargs.items()]
52        if len(params) > 0:
53            url += "?" + "&".join(params)
54        response = api.get(url)
55        return response