feat: Support webhooks API (#135)

* Add webhooks API

* Test webhooks

* Fix docstrings

Author: andrii-balitskyi

seamapi/routes.py

@@ -9,6 +9,7 @@
 from .access_codes import AccessCodes
 from .action_attempts import ActionAttempts
 from .thermostats import Thermostats
+from .webhooks import Webhooks
 
 class Routes(AbstractRoutes):
     def __init__(self):
@@ -22,6 +23,7 @@ def __init__(self):
       self.action_attempts = ActionAttempts(seam=self)
       self.noise_sensors = NoiseSensors(seam=self)
       self.thermostats = Thermostats(seam=self)
+      self.webhooks = Webhooks(seam=self)
 
     def make_request(self):
       raise NotImplementedError()

seamapi/types.py

@@ -17,6 +17,7 @@
 DeviceType = str  # e.g. august_lock
 WorkspaceId = str
 ClimateSettingScheduleId = str
+WebhookId = str
 
 
 class SeamApiException(Exception):
@@ -297,6 +298,15 @@ class ClimateSettingScheduleUpdate(ClimateSettingSchedule):
     pass
 
 
+@dataclass_json
+@dataclass
+class Webhook:
+    webhook_id: str
+    url: str
+    event_types: List[str] = None
+    secret: str = None
+
+
 class AbstractActionAttempts(abc.ABC):
     @abc.abstractmethod
     def get(
@@ -776,6 +786,34 @@ def set_fan_mode(
         raise NotImplementedError
 
 
+class AbstractWebhooks(abc.ABC):
+    @abc.abstractmethod
+    def create(
+        self,
+        url: str,
+        event_types: Optional[list] = None,
+    ) -> Webhook:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def delete(
+        self,
+        webhook: Union[WebhookId, Webhook],
+    ) -> bool:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def get(
+        self,
+        webhook: Union[WebhookId, Webhook],
+    ) -> Webhook:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def list(self) -> List[Webhook]:
+        raise NotImplementedError
+
+
 @dataclass
 class AbstractRoutes(abc.ABC):
     workspaces: AbstractWorkspaces
@@ -788,6 +826,7 @@ class AbstractRoutes(abc.ABC):
     thermostats: AbstractThermostats
     events: AbstractEvents
     connected_accounts: AbstractConnectedAccounts
+    webhooks: AbstractWebhooks
 
     @abc.abstractmethod
     def make_request(self, method: str, path: str, **kwargs) -> Any:

seamapi/utils/convert_to_id.py

@@ -11,6 +11,8 @@
     ConnectedAccountId,
     Device,
     DeviceId,
+    Webhook,
+    WebhookId,
     Workspace,
     WorkspaceId,
     ClimateSettingSchedule,
@@ -31,15 +33,16 @@ def to_device_id(device: Union[DeviceId, Device]) -> str:
         return device
     return device.device_id
 
-def to_climate_setting_schedule_id(climate_setting_schedule: Union[ClimateSettingScheduleId, ClimateSettingSchedule]) -> str:
+
+def to_climate_setting_schedule_id(
+    climate_setting_schedule: Union[ClimateSettingScheduleId, ClimateSettingSchedule]
+) -> str:
     if isinstance(climate_setting_schedule, str):
         return climate_setting_schedule
     return climate_setting_schedule.climate_setting_schedule_id
 
 
-def to_action_attempt_id(
-    action_attempt: Union[ActionAttemptId, ActionAttempt]
-) -> str:
+def to_action_attempt_id(action_attempt: Union[ActionAttemptId, ActionAttempt]) -> str:
     if isinstance(action_attempt, str):
         return action_attempt
     return action_attempt.action_attempt_id
@@ -71,3 +74,9 @@ def to_event_id(event: Union[EventId, Event]) -> str:
     if isinstance(event, str):
         return event
     return event.event_id
+
+
+def to_webhook_id(webhook: Union[WebhookId, Webhook]) -> str:
+    if isinstance(webhook, str):
+        return webhook
+    return webhook.webhook_id

seamapi/webhooks.py

@@ -0,0 +1,171 @@
+from seamapi.types import (
+    AbstractSeam as Seam,
+    AbstractWebhooks,
+    Webhook,
+    WebhookId,
+)
+import time
+from typing import List, Union, Optional, cast
+import requests
+from seamapi.utils.convert_to_id import (
+    to_connect_webview_id,
+    to_connected_account_id,
+    to_device_id,
+    to_webhook_id,
+)
+from seamapi.utils.report_error import report_error
+
+
+class Webhooks(AbstractWebhooks):
+    """
+    A class used to interact with webhooks API
+
+    ...
+
+    Attributes
+    ----------
+    seam : Seam
+        Initial seam class
+
+    Methods
+    -------
+    create(url, event_types=None)
+        Creates a new webhook
+    delete(webhook_id)
+        Deletes a webhook
+    get(webhook_id)
+        Fetches a webhook
+    list()
+        Lists webhooks
+    """
+
+    seam: Seam
+
+    def __init__(self, seam: Seam):
+        """
+        Parameters
+        ----------
+        seam : Seam
+          Initial seam class
+        """
+
+        self.seam = seam
+
+    @report_error
+    def create(
+        self,
+        url: str,
+        event_types: Optional[list] = None,
+    ) -> Webhook:
+        """Creates a new webhook.
+
+        Parameters
+        ----------
+        url : str
+            URL to send webhook events to
+        event_types : Optional[List[str]]
+            List of event types to send to webhook eg. ["connected_account.connected"]. Defaults to ["*"]
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            A webhook.
+        """
+        create_payload = {"url": url}
+        if event_types is not None:
+            create_payload["event_types"] = event_types
+
+        res = self.seam.make_request(
+            "POST",
+            "/webhooks/create",
+            json=create_payload,
+        )
+
+        return Webhook.from_dict(res["webhook"])
+
+    @report_error
+    def delete(
+        self,
+        webhook: Union[WebhookId, Webhook],
+    ) -> bool:
+        """Deletes a webhook.
+
+        Parameters
+        ----------
+        webhook : Union[WebhookId, Webhook]
+            Webhook ID or Webhook
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            Boolean.
+        """
+
+        res = self.seam.make_request(
+            "DELETE",
+            "/webhooks/delete",
+            json={"webhook_id": to_webhook_id(webhook)},
+        )
+
+        return True
+
+    @report_error
+    def get(
+        self,
+        webhook: Union[WebhookId, Webhook],
+    ) -> Webhook:
+        """Fetches a webhook.
+
+        Parameters
+        ----------
+        webhook : Union[WebhookId, Webhook]
+            Webhook ID or Webhook
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            A webhook.
+        """
+
+        res = self.seam.make_request(
+            "GET",
+            "/webhooks/get",
+            params={"webhook_id": to_webhook_id(webhook)},
+        )
+
+        return Webhook.from_dict(res["webhook"])
+
+    @report_error
+    def list(
+        self,
+    ) -> List[Webhook]:
+        """Lists webhooks.
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            A list of webhooks.
+        """
+
+        res = self.seam.make_request(
+            "GET",
+            "/webhooks/list",
+        )
+
+        return [Webhook.from_dict(w) for w in res["webhooks"]]

tests/webhooks/test_webhooks.py

@@ -0,0 +1,18 @@
+from seamapi import Seam
+
+
+def test_workspaces(seam: Seam):
+    webhook = seam.webhooks.create(
+        url="https://example.com", event_types=["connected_account.connected"]
+    )
+    assert webhook.url == "https://example.com"
+
+    webhook = seam.webhooks.get(webhook)
+    assert webhook is not None
+
+    webhook_list = seam.webhooks.list()
+    assert len(webhook_list) > 0
+
+    seam.webhooks.delete(webhook)
+    webhook_list = seam.webhooks.list()
+    assert len(webhook_list) == 0