Webhooks

Contains UP42 webhooks functionality to set up a custom callback e.g. when an order is finished he webhook is triggered and an event notification is transmitted via HTTPS to a specific URL.

Also see the full webhook documentation.

Create a new webhook or query a existing ones via the up42 object, e.g.

webhooks = up42.get_webhooks()

webhook = up42.initialize_webhook(webhook_id = "...")

The resulting Webhook object lets you modify, test or delete the specific webhook, e.g.

webhook = webhook.trigger_test_event()

Source code in up42/webhooks.py

class Webhooks:
    """
    Contains UP42 webhooks functionality to set up a custom callback e.g. when an order is finished
    he webhook is triggered and an event notification is transmitted via HTTPS to a specific URL.

    Also see the [full webhook documentation](https://docs.up42.com/account/webhooks).

    Create a new webhook or query a existing ones via the `up42` object, e.g.
    ```python
    webhooks = up42.get_webhooks()
    ```
    ```python
    webhook = up42.initialize_webhook(webhook_id = "...")
    ```

    The resulting Webhook object lets you modify, test or delete the specific webhook, e.g.
    ```python
    webhook = webhook.trigger_test_event()
    ```
    """

    def __init__(self, auth: Auth):
        self.auth = auth
        self.workspace_id = auth.workspace_id

    def get_webhook_events(self) -> dict:
        """
        Gets all available webhook events.

        Returns:
            A dict of the available webhook events.
        """
        url = f"{self.auth._endpoint()}/webhooks/events"
        response_json = self.auth._request(request_type="GET", url=url)
        return response_json["data"]

    def get_webhooks(self, return_json: bool = False) -> List[Webhook]:
        """
        Gets all registered webhooks for this workspace.

        Args:
            return_json: If true returns the webhooks information as JSON instead of webhook class objects.

        Returns:
            A list of the registered webhooks for this workspace.
        """
        url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks"
        response_json = self.auth._request(request_type="GET", url=url)
        logger.info(f"Queried {len(response_json['data'])} webhooks.")

        if return_json:
            return response_json["data"]
        webhooks = [
            Webhook(
                auth=self.auth, webhook_id=webhook_info["id"], webhook_info=webhook_info
            )
            for webhook_info in response_json["data"]
        ]
        return webhooks

    def create_webhook(
        self,
        name: str,
        url: str,
        events: List[str],
        active: bool = False,
        secret: Optional[str] = None,
    ) -> Webhook:
        """
        Registers a new webhook in the system.

        Args:
            name: Webhook name
            url: Unique URL where the webhook will send the message (HTTPS required)
            events: List of event types e.g. [order.status, job.status]
            active: Webhook status.
            secret: String that acts as signature to the https request sent to the url.

        Returns:
            A dict with details of the registered webhook.
        """
        input_parameters = {
            "name": name,
            "url": url,
            "events": events,
            "secret": secret,
            "active": active,
        }
        url_post = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks"
        response_json = self.auth._request(
            request_type="POST", url=url_post, data=input_parameters
        )
        webhook = Webhook(
            auth=self.auth,
            webhook_id=response_json["data"]["id"],
            webhook_info=response_json["data"],
        )
        logger.info(f"Created webhook {webhook}")
        return webhook

Functions

create_webhook(name, url, events, active=False, secret=None)

Registers a new webhook in the system.

Parameters:

Name	Type	Description	Default
name	str	Webhook name	required
url	str	Unique URL where the webhook will send the message (HTTPS required)	required
events	List[str]	List of event types e.g. [order.status, job.status]	required
active	bool	Webhook status.	False
secret	Optional[str]	String that acts as signature to the https request sent to the url.	None

Returns:

Type	Description
Webhook	A dict with details of the registered webhook.

Source code in up42/webhooks.py

def create_webhook(
    self,
    name: str,
    url: str,
    events: List[str],
    active: bool = False,
    secret: Optional[str] = None,
) -> Webhook:
    """
    Registers a new webhook in the system.

    Args:
        name: Webhook name
        url: Unique URL where the webhook will send the message (HTTPS required)
        events: List of event types e.g. [order.status, job.status]
        active: Webhook status.
        secret: String that acts as signature to the https request sent to the url.

    Returns:
        A dict with details of the registered webhook.
    """
    input_parameters = {
        "name": name,
        "url": url,
        "events": events,
        "secret": secret,
        "active": active,
    }
    url_post = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks"
    response_json = self.auth._request(
        request_type="POST", url=url_post, data=input_parameters
    )
    webhook = Webhook(
        auth=self.auth,
        webhook_id=response_json["data"]["id"],
        webhook_info=response_json["data"],
    )
    logger.info(f"Created webhook {webhook}")
    return webhook

get_webhook_events()

Gets all available webhook events.

Returns:

Type	Description
dict	A dict of the available webhook events.

Source code in up42/webhooks.py

def get_webhook_events(self) -> dict:
    """
    Gets all available webhook events.

    Returns:
        A dict of the available webhook events.
    """
    url = f"{self.auth._endpoint()}/webhooks/events"
    response_json = self.auth._request(request_type="GET", url=url)
    return response_json["data"]

get_webhooks(return_json=False)

Gets all registered webhooks for this workspace.

Parameters:

Name	Type	Description	Default
return_json	bool	If true returns the webhooks information as JSON instead of webhook class objects.	False

Returns:

Type	Description
List[Webhook]	A list of the registered webhooks for this workspace.

Source code in up42/webhooks.py

def get_webhooks(self, return_json: bool = False) -> List[Webhook]:
    """
    Gets all registered webhooks for this workspace.

    Args:
        return_json: If true returns the webhooks information as JSON instead of webhook class objects.

    Returns:
        A list of the registered webhooks for this workspace.
    """
    url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks"
    response_json = self.auth._request(request_type="GET", url=url)
    logger.info(f"Queried {len(response_json['data'])} webhooks.")

    if return_json:
        return response_json["data"]
    webhooks = [
        Webhook(
            auth=self.auth, webhook_id=webhook_info["id"], webhook_info=webhook_info
        )
        for webhook_info in response_json["data"]
    ]
    return webhooks

Webhook

Webhook class to control a specific UP42 webhook, e.g. modify, test or delete the specific webhook.

webhook = webhook.trigger_test_event()

Source code in up42/webhooks.py

class Webhook:
    """
    # Webhook

    Webhook class to control a specific UP42 webhook, e.g. modify, test or delete the specific webhook.

    ```python
    webhook = webhook.trigger_test_event()
    ```
    """

    def __init__(self, auth: Auth, webhook_id: str, webhook_info: dict = None):
        self.auth = auth
        self.workspace_id = auth.workspace_id
        self.webhook_id = webhook_id
        if webhook_info is not None:
            self._info = webhook_info
        else:
            self._info = self.info

    def __repr__(self):
        return f"Webhook(name: {self._info['name']}, webhook_id: {self.webhook_id}, active: {self._info['active']}"

    @property
    def info(self) -> dict:
        """
        Gets and updates the webhook metadata information.
        """
        url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}"
        response_json = self.auth._request(request_type="GET", url=url)
        self._info = response_json["data"]
        return self._info

    def trigger_test_events(self) -> dict:
        """
        Triggers webhook test event to test your receiving side. The UP42 server will send test
        messages for each subscribed event to the specified webhook URL.

        Returns:
            A dict with information about the test events.
        """
        url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}/tests"
        response_json = self.auth._request(
            request_type="POST",
            url=url,
        )
        return response_json["data"]

    def update(
        self,
        name: Optional[str] = None,
        url: Optional[str] = None,
        events: Optional[List[str]] = None,
        active: Optional[bool] = None,
        secret: Optional[str] = None,
    ) -> "Webhook":
        """
        Updates a registered webhook.

        Args:
            name: Updated webhook name
            url: Updated unique URL where the webhook will send the message (HTTPS required)
            events: Updated list of event types [order.status, job.status].
            active: Updated webhook status.
            secret: Updated string that acts as signature to the https request sent to the url.

        Returns:
            The updated webhook object.
        """
        self.info  # _info could be outdated. #pylint: disable=pointless-statement
        input_parameters = {
            "name": name if name is not None else self._info["name"],
            "url": url if url is not None else self._info["url"],
            "events": events if events is not None else self._info["events"],
            "secret": secret if secret is not None else self._info["secret"],
            "active": active if active is not None else self._info["active"],
        }
        url_put = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}"
        response_json = self.auth._request(
            request_type="PUT", url=url_put, data=input_parameters
        )
        self._info = response_json["data"]
        logger.info(f"Updated webhook {self}")
        return self

    def delete(self) -> None:
        """
        Deletes a registered webhook.
        """
        url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}"
        self.auth._request(request_type="DELETE", url=url)
        logger.info(f"Successfully deleted Webhook: {self.webhook_id}")

Attributes

info: dict property

Gets and updates the webhook metadata information.

Functions

delete()

Deletes a registered webhook.

Source code in up42/webhooks.py

def delete(self) -> None:
    """
    Deletes a registered webhook.
    """
    url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}"
    self.auth._request(request_type="DELETE", url=url)
    logger.info(f"Successfully deleted Webhook: {self.webhook_id}")

trigger_test_events()

Triggers webhook test event to test your receiving side. The UP42 server will send test messages for each subscribed event to the specified webhook URL.

Returns:

Type	Description
dict	A dict with information about the test events.

Source code in up42/webhooks.py

def trigger_test_events(self) -> dict:
    """
    Triggers webhook test event to test your receiving side. The UP42 server will send test
    messages for each subscribed event to the specified webhook URL.

    Returns:
        A dict with information about the test events.
    """
    url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}/tests"
    response_json = self.auth._request(
        request_type="POST",
        url=url,
    )
    return response_json["data"]

update(name=None, url=None, events=None, active=None, secret=None)

Updates a registered webhook.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Updated webhook name	None
url	Optional[str]	Updated unique URL where the webhook will send the message (HTTPS required)	None
events	Optional[List[str]]	Updated list of event types [order.status, job.status].	None
active	Optional[bool]	Updated webhook status.	None
secret	Optional[str]	Updated string that acts as signature to the https request sent to the url.	None

Returns:

Type	Description
Webhook	The updated webhook object.

Source code in up42/webhooks.py

def update(
    self,
    name: Optional[str] = None,
    url: Optional[str] = None,
    events: Optional[List[str]] = None,
    active: Optional[bool] = None,
    secret: Optional[str] = None,
) -> "Webhook":
    """
    Updates a registered webhook.

    Args:
        name: Updated webhook name
        url: Updated unique URL where the webhook will send the message (HTTPS required)
        events: Updated list of event types [order.status, job.status].
        active: Updated webhook status.
        secret: Updated string that acts as signature to the https request sent to the url.

    Returns:
        The updated webhook object.
    """
    self.info  # _info could be outdated. #pylint: disable=pointless-statement
    input_parameters = {
        "name": name if name is not None else self._info["name"],
        "url": url if url is not None else self._info["url"],
        "events": events if events is not None else self._info["events"],
        "secret": secret if secret is not None else self._info["secret"],
        "active": active if active is not None else self._info["active"],
    }
    url_put = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/webhooks/{self.webhook_id}"
    response_json = self.auth._request(
        request_type="PUT", url=url_put, data=input_parameters
    )
    self._info = response_json["data"]
    logger.info(f"Updated webhook {self}")
    return self