Add access_code_ids param to access_codes.list and type param to access_codes.update (#113)

Author: andrii-balitskyi

seamapi/access_codes.py

@@ -53,13 +53,21 @@ def __init__(self, seam: Seam):
         self.seam = seam
 
     @report_error
-    def list(self, device: Union[DeviceId, Device]) -> List[AccessCode]:
+    def list(
+        self,
+        device: Optional[Union[DeviceId, Device]] = None,
+        access_codes: Optional[
+            Union[List[AccessCode], List[AccessCodeId]]
+        ] = None,
+    ) -> List[AccessCode]:
         """Gets a list of access codes for a device.
 
         Parameters
         ----------
-        device : DeviceId or Device
+        device : Union[DeviceId, Device], optional
             Device id or Device to list access codes for
+        access_codes : Union[List[AccessCode], List[AccessCodeId]], optional
+            Access Code IDs or Access Codes to filter access_codes by
 
         Raises
         ------
@@ -71,15 +79,22 @@ def list(self, device: Union[DeviceId, Device]) -> List[AccessCode]:
             A list of access codes for a device.
         """
 
-        device_id = to_device_id(device)
+        params = {}
+        if device:
+            params["device_id"] = to_device_id(device)
+        if access_codes:
+            params["access_code_ids"] = [
+                to_access_code_id(ac) for ac in access_codes
+            ]
+
         res = self.seam.make_request(
             "GET",
             "/access_codes/list",
-            params={"device_id": device_id},
+            params=params,
         )
-        access_codes = res["access_codes"]
+        res_access_codes = res["access_codes"]
 
-        return [AccessCode.from_dict(ac) for ac in access_codes]
+        return [AccessCode.from_dict(ac) for ac in res_access_codes]
 
     @report_error
     def get(
@@ -180,13 +195,19 @@ def create(
         if common_code_key is not None:
             create_payload["common_code_key"] = common_code_key
         if attempt_for_offline_device is not None:
-            create_payload["attempt_for_offline_device"] = attempt_for_offline_device
+            create_payload[
+                "attempt_for_offline_device"
+            ] = attempt_for_offline_device
 
-        if (wait_for_code
+        if (
+            wait_for_code
             and starts_at is not None
-            and datetime.fromisoformat(starts_at) > datetime.now() + timedelta(seconds=5)
+            and datetime.fromisoformat(starts_at)
+            > datetime.now() + timedelta(seconds=5)
         ):
-            raise RuntimeError("Cannot use wait_for_code with a future time bound code")
+            raise RuntimeError(
+                "Cannot use wait_for_code with a future time bound code"
+            )
 
         res = self.seam.make_request(
             "POST",
@@ -199,25 +220,25 @@ def create(
         duration = 0
         poll_interval = 0.25
         if wait_for_code:
-            while (access_code.code is None):
-                if (access_code.status == "unknown"):
+            while access_code.code is None:
+                if access_code.status == "unknown":
                     raise WaitForAccessCodeFailedException(
                         "Access code status returned unknown",
-                        access_code_id=access_code.access_code_id
+                        access_code_id=access_code.access_code_id,
                     )
-                if (len(access_code.errors) > 0):
+                if len(access_code.errors) > 0:
                     raise WaitForAccessCodeFailedException(
                         "Access code returned errors",
                         access_code_id=access_code.access_code_id,
-                        errors=access_code.errors
+                        errors=access_code.errors,
                     )
                 time.sleep(poll_interval)
                 duration += poll_interval
-                if (duration > timeout):
+                if duration > timeout:
                     raise WaitForAccessCodeFailedException(
                         f"Gave up after waiting the maximum timeout of {timeout} seconds",
                         access_code_id=access_code.access_code_id,
-                        errors=access_code.errors
+                        errors=access_code.errors,
                     )
 
                 access_code = access_codes.get(access_code)
@@ -294,6 +315,7 @@ def update(
         code: Optional[str] = None,
         starts_at: Optional[str] = None,
         ends_at: Optional[str] = None,
+        type: Optional[str] = None,
     ) -> AccessCode:
         """Updates an access code on a device.
 
@@ -311,6 +333,8 @@ def update(
             Time when access code becomes effective
         ends_at : str, optional
             Time when access code ceases to be effective
+        type : str, optional
+            Access code type eg. ongoing or time_bound
 
         Raises
         ------
@@ -334,6 +358,8 @@ def update(
             update_payload["starts_at"] = starts_at
         if ends_at is not None:
             update_payload["ends_at"] = ends_at
+        if type is not None:
+            update_payload["type"] = type
 
         res = self.seam.make_request(
             "POST",

seamapi/types.py

@@ -293,7 +293,13 @@ def unlock_door(self, device: Union[DeviceId, Device]) -> ActionAttempt:
 
 class AbstractAccessCodes(abc.ABC):
     @abc.abstractmethod
-    def list(self, device: Union[DeviceId, Device]) -> List[AccessCode]:
+    def list(
+        self,
+        device: Optional[Union[DeviceId, Device]] = None,
+        access_codes: Optional[
+            Union[List[AccessCode], List[AccessCodeId]]
+        ] = None,
+    ) -> List[AccessCode]:
         raise NotImplementedError
 
     @abc.abstractmethod
@@ -336,7 +342,7 @@ def update(
         code: Optional[str] = None,
         starts_at: Optional[str] = None,
         ends_at: Optional[str] = None,
-        status: Optional[str] = None,
+        type: Optional[str] = None,
     ) -> AccessCode:
         raise NotImplementedError
 

tests/access_codes/test_access_codes.py

@@ -16,18 +16,34 @@ def test_access_codes(seam: Seam):
     assert created_access_code.name == "Test code"
     assert created_access_code.status == "setting"
 
+    seam.access_codes.create(some_device.device_id, "Test code 2", "5555")
+
     access_codes = seam.access_codes.list(some_device.device_id)
+    assert len(access_codes) == 2
+    access_codes = seam.access_codes.list(
+        some_device, access_codes=[created_access_code]
+    )
     assert len(access_codes) == 1
 
     access_code = seam.access_codes.get(created_access_code.access_code_id)
     assert access_code.code == "4444"
 
     with pytest.raises(SeamAPIException):
-        seam.access_codes.create(some_device.device_id, "Duplicate Access Code", "4444")
+        seam.access_codes.create(
+            some_device.device_id, "Duplicate Access Code", "4444"
+        )
 
     access_code = seam.access_codes.update(access_code, name="Updated name")
     assert access_code.name == "Updated name"
 
+    access_code = seam.access_codes.update(
+        access_code,
+        type="time_bound",
+        starts_at="3001-01-01",
+        ends_at="3001-01-03",
+    )
+    assert access_code.type == "time_bound"
+
     delete_action_attempt = seam.access_codes.delete(created_access_code)
     assert delete_action_attempt.status == "success"