feat: Create Access Codes against many devices with same code (#60)

Author: seveibar

.github/CODEOWNERS

@@ -0,0 +1 @@
+* @codetheweb @seveibar @andrii-balitskyi

seamapi/access_codes.py

@@ -125,6 +125,7 @@ def create(
         code: Optional[str] = None,
         starts_at: Optional[str] = None,
         ends_at: Optional[str] = None,
+        common_code_key: Optional[str] = None,
     ) -> AccessCode:
         """Creates an access code on a device.
 
@@ -161,6 +162,8 @@ def create(
             create_payload["starts_at"] = starts_at
         if ends_at is not None:
             create_payload["ends_at"] = ends_at
+        if common_code_key is not None:
+            create_payload["common_code_key"] = common_code_key
 
         res = self.seam.make_request(
             "POST",
@@ -175,6 +178,67 @@ def create(
 
         return AccessCode.from_dict(success_res["access_code"])
 
+    @report_error
+    def create_multiple(
+        self,
+        devices: Union[List[DeviceId], List[Device]],
+        name: Optional[str] = None,
+        code: Optional[str] = None,
+        starts_at: Optional[str] = None,
+        ends_at: Optional[str] = None,
+    ) -> List[AccessCode]:
+        """Creates multiple access codes across multiple devices. All access
+        codes will have the same code (if possible).
+
+        Parameters
+        ----------
+        devices : List of DeviceIds or Devices
+            Device ids or Devices to create an access code for
+        name : str, optional
+            Access code name
+        code : str, optional
+            Access code value
+        starts_at : str, optional
+            Time when access code becomes effective
+        ends_at : str, optional
+            Time when access code ceases to be effective
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            AccessCode
+        """
+
+        device_ids: List[str] = []
+        for device in devices:
+            device_ids.append(to_device_id(device))
+
+        create_payload: dict[str, Any] = {"device_ids": device_ids}
+        if name is not None:
+            create_payload["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 = self.seam.make_request(
+            "POST",
+            "/access_codes/create_multiple",
+            json=create_payload,
+        )
+
+        access_codes: List[AccessCode] = []
+        for access_code in res["access_codes"]:
+            access_codes.append(AccessCode.from_dict(access_code))
+
+        return access_codes
+
     @report_error
     def update(
         self,

seamapi/types.py

@@ -16,13 +16,19 @@
 DeviceType = str  # e.g. august_lock
 WorkspaceId = str
 
+
 class SeamAPIException(Exception):
-    def __init__(self, status_code: int, request_id: str, metadata: Optional[Dict[str, any]]):
+    def __init__(
+        self, status_code: int, request_id: str, metadata: Optional[Dict[str, any]]
+    ):
         self.status_code = status_code
         self.request_id = request_id
         self.metadata = metadata
 
-        super().__init__(f"SeamAPIException: status={status_code}, request_id={request_id}, metadata={metadata}")
+        super().__init__(
+            f"SeamAPIException: status={status_code}, request_id={request_id}, metadata={metadata}"
+        )
+
 
 class ActionAttemptFailedException(Exception):
     def __init__(
@@ -61,6 +67,7 @@ def from_dict(d: Dict[str, Any]):
             errors=d["errors"],
         )
 
+
 @dataclass
 class Event:
     event_id: str
@@ -75,6 +82,7 @@ class Event:
     end_time: Union[str, None]
     created_at: Union[str, None]
 
+
 @dataclass_json
 @dataclass
 class ActionAttemptError:
@@ -140,6 +148,7 @@ class AccessCode:
     ends_at: Optional[str] = None
     name: Optional[str] = ""
     status: Optional[str] = None
+    common_code_key: Optional[str] = None
 
 
 class AbstractActionAttempts(abc.ABC):
@@ -188,9 +197,27 @@ def get(
     ) -> AccessCode:
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def create_multiple(
+        self,
+        devices: Union[List[DeviceId], List[Device]],
+        name: Optional[str] = None,
+        code: Optional[str] = None,
+        starts_at: Optional[str] = None,
+        ends_at: Optional[str] = None,
+        common_code_key: Optional[str] = None,
+    ) -> List[AccessCode]:
+        raise NotImplementedError
+
     @abc.abstractmethod
     def create(
-        self, device: Union[DeviceId, Device], name: str, code: str
+        self,
+        device: Union[DeviceId, Device],
+        name: Optional[str] = None,
+        code: Optional[str] = None,
+        starts_at: Optional[str] = None,
+        ends_at: Optional[str] = None,
+        common_code_key: Optional[str] = None,
     ) -> AccessCode:
         raise NotImplementedError
 
@@ -228,11 +255,13 @@ def get(
     ) -> Device:
         raise NotImplementedError
 
+
 class AbstractEvents(abc.ABC):
     @abc.abstractmethod
     def list(self) -> List[Event]:
         raise NotImplementedError
 
+
 class AbstractWorkspaces(abc.ABC):
     @abc.abstractmethod
     def list(self) -> List[Workspace]:
@@ -274,6 +303,7 @@ def get(
     ) -> ConnectedAccount:
         raise NotImplementedError
 
+
 @dataclass
 class AbstractRoutes(abc.ABC):
     workspaces: AbstractWorkspaces
@@ -297,6 +327,7 @@ class AbstractSeam(AbstractRoutes):
     def __init__(self, api_key: Optional[str] = None):
         raise NotImplementedError
 
+
 @dataclass_json
 @dataclass
 class ResetSandBoxResponse:

tests/access_codes/test_access_codes.py

@@ -7,7 +7,8 @@
 def test_access_codes(seam: Seam):
     run_august_factory(seam)
 
-    some_device = seam.devices.list()[0]
+    all_devices = seam.devices.list()
+    some_device = all_devices[0]
 
     created_access_code = seam.access_codes.create(
         some_device.device_id, "Test code", "4444"
@@ -29,3 +30,7 @@ def test_access_codes(seam: Seam):
 
     delete_action_attempt = seam.access_codes.delete(created_access_code)
     assert delete_action_attempt.status == "success"
+
+    access_codes = seam.access_codes.create_multiple(devices=all_devices)
+    assert len(access_codes) == len(all_devices)
+    assert len(set([ac.common_code_key for ac in access_codes])) == 1