action attempts implementation initial

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "seamapi"
-version = "0.2.2"
+version = "0.3.0"
 description = "A Python Library for Seam's API https://getseam.com"
 authors = ["Severin Ibarluzea <seveibar@gmail.com>"]
 license = "MIT"

seamapi/access_codes.py

@@ -1,14 +1,22 @@
 from seamapi.types import (
     AbstractAccessCodes,
     AccessCode,
+    AccessCodeId,
+    ActionAttempt,
     Device,
     DeviceId,
     AbstractSeam as Seam,
 )
-from typing import List, Union
+from typing import List, Optional, Union, Any
 import requests
 
 
+def to_access_code_id(access_code: Union[AccessCodeId, AccessCode]) -> str:
+    if isinstance(access_code, str):
+        return access_code
+    return access_code.access_code_id
+
+
 def to_device_id(device: Union[DeviceId, Device]) -> str:
     if isinstance(device, str):
         return device
@@ -32,16 +40,60 @@ def list(self, device: Union[DeviceId, Device]) -> List[AccessCode]:
         access_codes = res.json()["access_codes"]
         return [AccessCode.from_dict(ac) for ac in access_codes]
 
-    def create(self, device: Union[DeviceId, Device], name: str, code: str) -> None:
+    def get(self, access_code: Union[AccessCodeId, AccessCode]) -> AccessCode:
+        access_code_id = to_access_code_id(access_code)
+        res = requests.get(
+            f"{self.seam.api_url}/access_codes/get",
+            headers={"Authorization": f"Bearer {self.seam.api_key}"},
+            params={"access_code_id": access_code_id},
+        )
+        if not res.ok:
+            raise Exception(res.text)
+        return AccessCode.from_dict(res.json()["access_code"])
+
+    def create(
+        self,
+        device: Union[DeviceId, Device],
+        name: str,
+        code: Optional[str] = None,
+        starts_at: Optional[str] = None,
+        ends_at: Optional[str] = None,
+    ) -> AccessCode:
         device_id = to_device_id(device)
+        create_payload = {"device_id": device_id, "name": name}
+        if code is not None:
+            create_payload["code"] = code
+        if starts_at is not None:
+            create_payload["starts_at"] = starts_at
+        if ends_at is not None:
+            create_payload["ends_at"] = ends_at
         res = requests.post(
             f"{self.seam.api_url}/access_codes/create",
             headers={"Authorization": f"Bearer {self.seam.api_key}"},
-            json={
-                "device_id": device_id,
-                "name": name,
-                "code": code,
-            },
+            json=create_payload,
         )
         if not res.ok:
             raise Exception(res.text)
+        action_attempt = self.seam.action_attempts.poll_until_ready(
+            res.json()["action_attempt"]["action_attempt_id"]
+        )
+        success_res: Any = action_attempt.result
+        return AccessCode.from_dict(success_res["access_code"])
+
+    def delete(self, access_code: Union[AccessCodeId, AccessCode]) -> ActionAttempt:
+        access_code_id = to_access_code_id(access_code)
+        res = requests.delete(
+            (f"{self.seam.api_url}/access_codes/delete"),
+            headers={"Authorization": f"Bearer {self.seam.api_key}"},
+            json={"access_code_id": access_code_id},
+        )
+        if not res.ok:
+            raise Exception(res.text)
+        action_attempt = self.seam.action_attempts.poll_until_ready(
+            res.json()["action_attempt"]["action_attempt_id"]
+        )
+        if action_attempt.status == "error" and action_attempt.error:
+            raise Exception(
+                f"{action_attempt.error.type}: {action_attempt.error.message}"
+            )
+        return action_attempt

seamapi/action_attempts.py

@@ -0,0 +1,60 @@
+from seamapi.types import (
+    AbstractActionAttempts,
+    ActionAttemptError,
+    ActionAttempt,
+    AbstractSeam as Seam,
+    ActionAttemptId,
+)
+import time
+import requests
+from typing import Union
+
+
+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
+
+
+class ActionAttempts(AbstractActionAttempts):
+    seam: Seam
+
+    def __init__(self, seam: Seam):
+        self.seam = seam
+
+    def get(
+        self, action_attempt: Union[ActionAttemptId, ActionAttempt]
+    ) -> ActionAttempt:
+        action_attempt_id = to_action_attempt_id(action_attempt)
+        res = requests.get(
+            f"{self.seam.api_url}/action_attempts/get",
+            headers={"Authorization": f"Bearer {self.seam.api_key}"},
+            params={"action_attempt_id": action_attempt_id},
+        )
+        if not res.ok:
+            raise Exception(res.text)
+        json_aa = res.json()["action_attempt"]
+        error = None
+        if "error" in json_aa and json_aa["error"] is not None:
+            error = ActionAttemptError(
+                type=json_aa["error"]["type"],
+                message=json_aa["error"]["message"],
+            )
+        return ActionAttempt(
+            action_attempt_id=json_aa["action_attempt_id"],
+            status=json_aa["status"],
+            action_type=json_aa["action_type"],
+            result=json_aa["result"],
+            error=error,
+        )
+
+    def poll_until_ready(
+        self, action_attempt: Union[ActionAttemptId, ActionAttempt]
+    ) -> ActionAttempt:
+        updated_action_attempt = None
+        while (
+            updated_action_attempt is None or updated_action_attempt.status == "pending"
+        ):
+            updated_action_attempt = self.get(action_attempt)
+            time.sleep(0.25)
+        return updated_action_attempt

seamapi/devices.py

@@ -3,14 +3,36 @@
 import requests
 
 
+def to_device_id(device: Union[DeviceId, Device]) -> str:
+    if isinstance(device, str):
+        return device
+    return device.device_id
+
+
 class Devices(AbstractDevices):
     seam: Seam
 
     def __init__(self, seam: Seam):
         self.seam = seam
 
     def list(self) -> List[Device]:
-        raise NotImplementedError
+        res = requests.get(
+            f"{self.seam.api_url}/devices/list",
+            headers={"Authorization": f"Bearer {self.seam.api_key}"},
+        )
+        if not res.ok:
+            raise Exception(res.text)
+        devices = res.json()["devices"]
+        return [Device.from_dict(d) for d in devices]
 
     def get(self, device: Union[DeviceId, Device]) -> Device:
-        raise NotImplementedError
+        device_id = to_device_id(device)
+        res = requests.get(
+            f"{self.seam.api_url}/devices/get",
+            headers={"Authorization": f"Bearer {self.seam.api_key}"},
+            params={"device_id": device_id},
+        )
+        if not res.ok:
+            raise Exception(res.text)
+        json_device = res.json()["device"]
+        return Device.from_dict(json_device)

seamapi/locks.py

@@ -1,10 +1,12 @@
 from seamapi.types import (
     AbstractLocks,
+    ActionAttempt,
     Device,
     DeviceId,
     AbstractSeam as Seam,
 )
-from typing import List, Union, Optional
+import time
+from typing import List, Union, Optional, cast
 import requests
 
 
@@ -43,7 +45,7 @@ def get(self, device: Union[DeviceId, Device]) -> Device:
         json_lock = res.json()["device"]
         return Device.from_dict(json_lock)
 
-    def lock_door(self, device: Union[DeviceId, Device]) -> None:
+    def lock_door(self, device: Union[DeviceId, Device]) -> ActionAttempt:
         device_id = to_device_id(device)
         res = requests.post(
             f"{self.seam.api_url}/locks/lock_door",
@@ -52,8 +54,11 @@ def lock_door(self, device: Union[DeviceId, Device]) -> None:
         )
         if not res.ok:
             raise Exception(res.text)
+        return self.seam.action_attempts.poll_until_ready(
+            res.json()["action_attempt"]["action_attempt_id"]
+        )
 
-    def unlock_door(self, device: Union[DeviceId, Device]) -> None:
+    def unlock_door(self, device: Union[DeviceId, Device]) -> ActionAttempt:
         device_id = to_device_id(device)
         res = requests.post(
             f"{self.seam.api_url}/locks/unlock_door",
@@ -62,3 +67,6 @@ def unlock_door(self, device: Union[DeviceId, Device]) -> None:
         )
         if not res.ok:
             raise Exception(res.text)
+        return self.seam.action_attempts.poll_until_ready(
+            res.json()["action_attempt"]["action_attempt_id"]
+        )

seamapi/seam.py

@@ -6,6 +6,7 @@
 from .connect_webviews import ConnectWebviews
 from .locks import Locks
 from .access_codes import AccessCodes
+from .action_attempts import ActionAttempts
 from .types import AbstractSeam
 
 
@@ -28,3 +29,4 @@ def __init__(self, api_key: Optional[str] = None):
         self.devices = Devices(seam=self)
         self.locks = Locks(seam=self)
         self.access_codes = AccessCodes(seam=self)
+        self.action_attempts = ActionAttempts(seam=self)

seamapi/types.py

@@ -5,6 +5,8 @@
 from dataclasses import dataclass
 from dataclasses_json import dataclass_json
 
+AccessCodeId = str
+ActionAttemptId = str
 DeviceId = str
 AcceptedProvider = str  # e.g. august or noiseaware
 
@@ -18,11 +20,21 @@ class Device:
     properties: Dict[str, Any]
 
 
+@dataclass_json
+@dataclass
+class ActionAttemptError:
+    type: str
+    message: str
+
+
 @dataclass_json
 @dataclass
 class ActionAttempt:
     action_attempt_id: str
+    action_type: str
     status: str
+    result: Optional[Any]
+    error: Optional[ActionAttemptError]
 
 
 @dataclass_json
@@ -58,9 +70,25 @@ class AccessCode:
     access_code_id: str
     type: str
     code: str
+    starts_at: Optional[str] = None
+    ends_at: Optional[str] = None
     name: Optional[str] = ""
 
 
+class AbstractActionAttempts(abc.ABC):
+    @abc.abstractmethod
+    def get(
+        self, action_attempt: Union[ActionAttemptId, ActionAttempt]
+    ) -> ActionAttempt:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def poll_until_ready(
+        self, action_attempt: Union[ActionAttemptId, ActionAttempt]
+    ) -> ActionAttempt:
+        raise NotImplementedError
+
+
 class AbstractLocks(abc.ABC):
     @abc.abstractmethod
     def list(self, connected_account: Optional[str] = None) -> List[Device]:
@@ -71,11 +99,11 @@ def get(self, device: Union[DeviceId, Device]) -> Device:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def lock_door(self, device: Union[DeviceId, Device]) -> None:
+    def lock_door(self, device: Union[DeviceId, Device]) -> ActionAttempt:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def unlock_door(self, device: Union[DeviceId, Device]) -> None:
+    def unlock_door(self, device: Union[DeviceId, Device]) -> ActionAttempt:
         raise NotImplementedError
 
 
@@ -85,17 +113,23 @@ def list(self, device: Union[DeviceId, Device]) -> List[AccessCode]:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def create(self, device: Union[DeviceId, Device], name: str, code: str) -> None:
+    def get(
+        self,
+        access_code: Union[AccessCodeId, AccessCode],
+    ) -> AccessCode:
         raise NotImplementedError
 
-
-class AbstractActionAttempt(abc.ABC):
     @abc.abstractmethod
-    def list(self) -> List[ActionAttempt]:
+    def create(
+        self, device: Union[DeviceId, Device], name: str, code: str
+    ) -> AccessCode:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def get(self, workspace_id: Optional[str] = None) -> ActionAttempt:
+    def delete(
+        self,
+        access_code: Union[AccessCodeId, AccessCode],
+    ) -> None:
         raise NotImplementedError
 
 
@@ -119,9 +153,7 @@ def get(self, workspace_id: Optional[str] = None) -> Workspace:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def reset_sandbox(
-        self, workspace_id: Optional[str] = None, sandbox_type: Optional[str] = None
-    ) -> None:
+    def reset_sandbox(self) -> None:
         raise NotImplementedError
 
 
@@ -161,6 +193,7 @@ class AbstractSeam(abc.ABC):
     locks: AbstractLocks
     devices: AbstractDevices
     access_codes: AbstractAccessCodes
+    action_attempts: AbstractActionAttempts
 
     @abc.abstractmethod
     def __init__(self, api_key: Optional[str] = None):

seamapi/workspaces.py

@@ -29,9 +29,7 @@ def get(self, workspace_id: Optional[str] = None) -> Workspace:
             is_sandbox=res_json["workspace"]["is_sandbox"],
         )
 
-    def reset_sandbox(
-        self, workspace_id: Optional[str] = None, sandbox_type: Optional[str] = None
-    ) -> None:
+    def reset_sandbox(self) -> None:
         res = requests.post(
             f"{self.seam.api_url}/workspaces/reset_sandbox",
             headers={"Authorization": f"Bearer {self.seam.api_key}"},