Support Audit Trail get endpoint

Author: Henry Chan

workos/__init__.py

@@ -5,4 +5,4 @@
 
 api_key = os.getenv("WORKOS_API_KEY")
 project_id = os.getenv("WORKOS_PROJECT_ID")
-base_api_url = "https://api.workos.com/"
+base_api_url = "http://localhost:7000/"

workos/audit_trail.py

@@ -1,10 +1,12 @@
 import workos
 from workos.exceptions import ConfigurationException
-from workos.utils.request import RequestHelper, REQUEST_METHOD_POST
+from workos.resources.event import WorkOSEvent
+from workos.utils.request import RequestHelper, REQUEST_METHOD_GET, REQUEST_METHOD_POST
 from workos.utils.validation import AUDIT_TRAIL_MODULE, validate_settings
 
 EVENTS_PATH = "events"
 METADATA_LIMIT = 50
+DEFAULT_EVENT_LIMIT = 10
 
 
 class AuditTrail(object):
@@ -51,7 +53,7 @@ def create_event(self, event, idempotency_key=None):
             dict: Response from WorkOS
         """
         if len(event.get("metadata", {})) > METADATA_LIMIT:
-            raise Exception(
+            raise ValueError(
                 "Number of metadata keys exceeds {}.".format(METADATA_LIMIT)
             )
 
@@ -66,3 +68,80 @@ def create_event(self, event, idempotency_key=None):
             headers=headers,
             token=workos.api_key,
         )
+
+    def get_events(
+        self,
+        before=None,
+        after=None,
+        limit=DEFAULT_EVENT_LIMIT,
+        group=None,
+        action=None,
+        action_type=None,
+        actor_name=None,
+        actor_id=None,
+        target_name=None,
+        target_id=None,
+        occurred_at=None,
+        occurred_at_gt=None,
+        occurred_at_gte=None,
+        occurred_at_lt=None,
+        occurred_at_lte=None,
+        search=None,
+    ):
+        if before and after:
+            raise ValueError("Specify either before or after")
+
+        params = {
+            "before": before,
+            "after": after,
+            "limit": limit,
+        }
+
+        if group:
+            params["group"] = list(group)
+
+        if action:
+            params["action"] = list(action)
+
+        if action_type:
+            params["action_type"] = list(action_type)
+
+        if actor_name:
+            params["actor_name"] = list(actor_name)
+
+        if actor_id:
+            params["actor_id"] = list(actor_id)
+
+        if target_name:
+            params["target_name"] = list(target_name)
+
+        if target_id:
+            params["target_id"] = list(target_id)
+
+        if occurred_at:
+            params["occurred_at"] = occurred_at
+        else:
+            if occurred_at_gte:
+                params["occurred_at_gte"] = occurred_at_gte
+            elif occurred_at_gt:
+                params["occurred_at_gt"] = occurred_at_gt
+
+            if occurred_at_lte:
+                params["occurred_at_lte"] = occurred_at_lte
+            elif occurred_at_lt:
+                params["occurred_at_lt"] = occurred_at_lt
+
+        if search:
+            params["search"] = search
+
+        response = self.request_helper.request(
+            EVENTS_PATH, method=REQUEST_METHOD_GET, params=params, token=workos.api_key,
+        )
+
+        events = [
+            WorkOSEvent.construct_from_response(data) for data in response["data"]
+        ]
+        before = response["listMetadata"]["before"]
+        after = response["listMetadata"]["after"]
+
+        return (events, before, after)

workos/resources/base.py

@@ -1,37 +1,36 @@
 class WorkOSBaseResource(object):
-  """Representation of a WorkOS Resource as returned through the API.
+    """Representation of a WorkOS Resource as returned through the API.
     
   Attributes:
       OBJECT_FIELDS (list): List of fields a Resource is comprised of.
   """
 
-  OBJECT_FIELDS = []
+    OBJECT_FIELDS = []
 
-  @classmethod
-  def construct_from_response(cls, response):
-    """Returns an instance of WorkOSBaseResource.
+    @classmethod
+    def construct_from_response(cls, response):
+        """Returns an instance of WorkOSBaseResource.
 
     Args:
         response (dict): Resource data from a WorkOS API response
 
     Returns:
         WorkOSBaseResource: Instance of a WorkOSBaseResource with OBJECT_FIELDS fields set
     """
-    print(response)
-    obj = cls()
-    for field in cls.OBJECT_FIELDS:
-      setattr(obj, field, response[field])
+        obj = cls()
+        for field in cls.OBJECT_FIELDS:
+            setattr(obj, field, response[field])
 
-    return obj
+        return obj
 
-  def to_dict(self):
-    """Returns a dict representation of the WorkOSBaseResource.
+    def to_dict(self):
+        """Returns a dict representation of the WorkOSBaseResource.
         
     Returns:
         dict: A dict representation of the WorkOSBaseResource
     """
-    obj_dict = {}
-    for field in self.OBJECT_FIELDS:
-      obj_dict[field] = getattr(self, field, None)
+        obj_dict = {}
+        for field in self.OBJECT_FIELDS:
+            obj_dict[field] = getattr(self, field, None)
 
-    return obj_dict
+        return obj_dict

workos/resources/event.py

@@ -0,0 +1,42 @@
+from workos.resources.base import WorkOSBaseResource
+from workos.resources.event_action import WorkOSEventAction
+
+
+class WorkOSEvent(WorkOSBaseResource):
+    """Representation of an Event as returned by WorkOS through the Audit Trail feature.
+    
+    Attributes:
+        OBJECT_FIELDS (list): List of fields a WorkOSEvent is comprised of.
+    """
+
+    OBJECT_FIELDS = [
+        "id",
+        "group",
+        "location",
+        "latitude",
+        "longitude",
+        "type",
+        "actor_name",
+        "actor_id",
+        "target_name",
+        "target_id",
+        "metadata",
+        "occurred_at",
+    ]
+
+    @classmethod
+    def construct_from_response(cls, response):
+        event = super(WorkOSEvent, cls).construct_from_response(response)
+
+        event_action = WorkOSEventAction.construct_from_response(response["action"])
+        event.action = event_action
+
+        return event
+
+    def to_dict(self):
+        event_dict = super(WorkOSEvent, self).to_dict()
+
+        event_action_dict = self.action.to_dict()
+        event_dict["action"] = event_action_dict
+
+        return event_dict

workos/resources/event_action.py

@@ -0,0 +1,15 @@
+from workos.resources.base import WorkOSBaseResource
+
+
+class WorkOSEventAction(WorkOSBaseResource):
+    """Representation of an Event Action as returned by WorkOS through the Audit Trail feature.
+    
+    Attributes:
+        OBJECT_FIELDS (list): List of fields a WorkOSEventAction is comprised of.
+    """
+
+    OBJECT_FIELDS = [
+        "id",
+        "name",
+        "project_id",
+    ]

workos/resources/sso.py

@@ -1,5 +1,6 @@
 from workos.resources.base import WorkOSBaseResource
 
+
 class WorkOSProfile(WorkOSBaseResource):
     """Representation of a User Profile as returned by WorkOS through the SSO feature.
     
@@ -14,4 +15,4 @@ class WorkOSProfile(WorkOSBaseResource):
         "last_name",
         "connection_type",
         "idp_id",
-    ]
+    ]