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

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