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
Methods¶
create_webhook(self, 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(self)
¶
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(self, 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[up42.webhooks.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
readonly
¶
Gets and updates the webhook metadata information.
Methods¶
delete(self)
¶
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(self)
¶
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(self, 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