feat: Merge pull request #47 from seamapi/implement-events

fix #33: Implement `event.list()` and `Event`

seamapi/events.py

@@ -0,0 +1,97 @@
+from seamapi.types import (
+    Event,
+    AbstractEvents,
+    AbstractSeam as Seam,
+)
+from typing import List, Optional
+import requests
+
+
+class Events(AbstractEvents):
+    """
+    A class to interact with events through the Seam API
+
+    ...
+
+    Attributes
+    ----------
+    seam : Seam
+        Initial seam class
+
+    Methods
+    -------
+    list(since, device_id=None, device_ids=None, event_type=None, event_types=None)
+        Gets a list of events
+    """
+
+    seam: Seam
+
+    def __init__(self, seam: Seam):
+        """
+        Parameters
+        ----------
+        seam : Seam
+          Intial seam class
+        """
+        self.seam = seam
+
+    def list(
+        self,
+        since: str,
+        device_id: Optional[str] = None,
+        device_ids: Optional[list] = None,
+        event_type: Optional[str] = None,
+        event_types: Optional[list] = None
+    ) -> List[Event]:
+        """Gets a list of events.
+
+        Parameters
+        ----------
+			since : str
+				ISO 8601 timestamp of the earliest event to return
+			device_id : Optional[str]
+				Device ID to filter events by
+			device_ids : Optional[list]
+				Device IDs to filter events by
+			event_type : Optional[str]
+				Event type to filter events by
+			event_types : Optional[list]
+				Event types to filter events by
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            A list of events.
+        """
+        if not since:
+            raise Exception("'since' is required")
+
+        params = {
+            "since": since,
+        }
+
+        arguments = {
+            "device_id": device_id,
+            "device_ids": device_ids,
+            "event_type": event_type,
+            "event_types": event_types
+        }
+
+        for name in arguments:
+            if arguments[name]:
+                params.update({name: arguments[name]})
+
+        res = requests.get(
+            f"{self.seam.api_url}/events/list",
+            params=params,
+            headers={"Authorization": f"Bearer {self.seam.api_key}"},
+        )
+        if not res.ok:
+            raise Exception(res.text)
+
+        events = res.json()["events"]
+        return events

seamapi/seam.py

@@ -2,6 +2,7 @@
 from typing import Optional, cast
 from .workspaces import Workspaces
 from .devices import Devices
+from .events import Events
 from .connected_accounts import ConnectedAccounts
 from .connect_webviews import ConnectWebviews
 from .locks import Locks
@@ -30,6 +31,8 @@ class Seam(AbstractSeam):
         Connect webviews class
     devices : Devices
         Devices class
+    events : Events
+        Events class
     locks : Locks
         Locks class
     access_codes : AccessCodes
@@ -69,6 +72,7 @@ def __init__(
         self.connected_accounts = ConnectedAccounts(seam=self)
         self.connect_webviews = ConnectWebviews(seam=self)
         self.devices = Devices(seam=self)
+        self.events = Events(seam=self)
         self.locks = Locks(seam=self)
         self.access_codes = AccessCodes(seam=self)
         self.action_attempts = ActionAttempts(seam=self)

seamapi/types.py

@@ -54,6 +54,19 @@ def from_dict(d: Dict[str, Any]):
             errors=d["errors"],
         )
 
+@dataclass
+class Event:
+    event_id: str
+    noiseaware_id: str
+    activity_zone_id: str
+    event_class: Union[str, None]
+    event_type: Union[str, None]
+    duration: Union[int, None]
+    local_end_time: Union[str, None]
+    local_start_time: Union[str, None]
+    start_time: Union[str, None]
+    end_time: Union[str, None]
+    created_at: Union[str, None]
 
 @dataclass_json
 @dataclass
@@ -198,6 +211,10 @@ def get(
     ) -> Device:
         raise NotImplementedError
 
+class AbstractEvents(abc.ABC):
+    @abc.abstractmethod
+    def list(self) -> List[Event]:
+        raise NotImplementedError
 
 class AbstractWorkspaces(abc.ABC):
     @abc.abstractmethod

tests/events/test_events.py

@@ -0,0 +1,11 @@
+from seamapi import Seam
+from tests.fixtures.run_august_factory import run_august_factory
+
+SINCE="2021-01-01T00:00:00.000Z"
+EVENT_TYPE = "device.connected"
+
+def test_events(seam: Seam):
+    run_august_factory(seam)
+
+    events = seam.events.list(since=SINCE)
+    assert events[0]['event_type'] == EVENT_TYPE