Simplify incoming event creation

Author: William Chrisp

publisher/drivers/checkov.py

@@ -18,13 +18,13 @@ def get_events(self, file_data: dict, repo_name: str) -> Union[Exception, Tuple[
                 events.append(GuardrailPassed(
                     aggregate_id = repo_name + "." + result["resource"],
                     guardrail_id = result["check_id"],
-                    time = current_time,
+                    timestamp = current_time,
                 ))
             for result in file_data["results"]["failed_checks"]:
                 events.append(GuardrailActivated(
                     aggregate_id = repo_name + "." + result["resource"],
                     guardrail_id = result["check_id"],
-                    time = current_time,
+                    timestamp = current_time,
                 ))
             return tuple(events)
         return Exception(f"Unable to read Checkov results from file")

publisher/drivers/open_policy_agent.py

@@ -19,13 +19,13 @@ def get_events(self, file_data: dict, repo_name: str) -> Union[Exception, Tuple[
                     events.append(GuardrailPassed(
                         aggregate_id = repo_name + "." + result["input"]["metadata"]["namespace"] + "."  + result["input"]["metadata"]["name"],
                         guardrail_id = result["query"],
-                        time = current_time,
+                        timestamp = current_time,
                     ))
                 else:
                     events.append(GuardrailActivated(
                         aggregate_id = repo_name + "." + result["input"]["metadata"]["namespace"] + "." + result["input"]["metadata"]["name"],
                         guardrail_id = result["query"],
-                        time = current_time,
+                        timestamp = current_time,
                     ))
             return tuple(events)
         return Exception(f"Unable to read Open Policy Agent results from file")

publisher/entities/guardrail.py

@@ -7,12 +7,12 @@
 @dataclass
 class GuardrailActivated(BaseEvent):
     guardrail_id: str
-    time: int
+    timestamp: int
     event_type: Literal["guardrail_activated"] = "guardrail_activated"
 
 
 @dataclass
 class GuardrailPassed(BaseEvent):
     guardrail_id: str
-    time: int
+    timestamp: int
     event_type: Literal["guardrail_passed"] = "guardrail_passed"

src/adapters/controller.py

@@ -1,19 +1,43 @@
-from typing import Any, List, Tuple, Union
+from typing import Dict, List, Tuple, Union
+from uuid import uuid4
 
-from src.entities.events import Event
+from src.entities.events import Event, EVENT_CLASSES
 from src.entities.metrics import Metric
 from src.usecases.event_functions import EVENT_FUNCTIONS
 
+def _convert_payload_to_event(
+    payload: Dict, aggregate_version: int
+) -> Event:
+    try:
+        event_type = payload["event_type"]
+        aggregate_id = payload["aggregate_id"]
+        event_payload = payload.copy()
+        for key in ["event_type", "aggregate_id"]:
+            event_payload.pop(key, None)
+        event_class, payload_class = EVENT_CLASSES[event_type]
+        return event_class(
+            aggregate_id=aggregate_id,
+            aggregate_version=aggregate_version + 1,
+            event_id=str(uuid4()),
+            payload=payload_class(**event_payload)
+        )
+    except Exception as err:
+        return err
 
 def handle_event(
-    payload: Any, aggregate_events: List[Event]
+    payload: Dict, aggregate_events: List[Event]
 ) -> Union[Exception, Tuple[Event, List[Metric]]]:
     if "event_type" in payload:
-        if payload["event_type"] in EVENT_FUNCTIONS:
-            event, metrics = EVENT_FUNCTIONS[payload["event_type"]](
-                payload, aggregate_events
-            )
-            return (event, metrics)
+        event_type = payload["event_type"]
+        if event_type in EVENT_CLASSES:
+            event = _convert_payload_to_event(payload, len(aggregate_events))
+            if event_type in EVENT_FUNCTIONS:
+                event, metrics = EVENT_FUNCTIONS[event_type](
+                    event, aggregate_events
+                )
+                return (event, metrics)
+            else:
+                return Exception(f"No event type function defined: {event_type}")
         else:
-            return Exception(f"Unknown event type {payload['event_type']}")
+            return Exception(f"Unknown event type {event_type}")
     return Exception("Malformed event with no event_type")

src/drivers/dynamo_event_sink_source.py

@@ -41,9 +41,6 @@ def store_events(self, events: List[Event]) -> Optional[Exception]:
                     }
                     batch.put_item(Item=value)
         except Exception as err:
-            logger.error(
-                f"Couldn't add to table type of error is {type(err)} AND reason is {err}"
-            )
             return err
 
 
@@ -56,28 +53,30 @@ def _sort_events(self, event: Event) -> int:
         return event.aggregate_version
 
     def get_events_for_aggregate(self, aggregate_id: str) -> List[Event]:
-        response = self.dynamo_db_table.query(
-            KeyConditionExpression=Key("aggregate_id").eq(aggregate_id)
-        )
-        logger.msg(f"responses {response}")
+        try:
+            response = self.dynamo_db_table.query(
+                KeyConditionExpression=Key("aggregate_id").eq(aggregate_id)
+            )
 
-        events: List[Event] = []
+            events: List[Event] = []
 
-        for item in response["Items"]:
-            event_type = item["event_type"]
-            if event_type in EVENT_CLASSES:
-                event_class, payload_class = EVENT_CLASSES[event_type]
-                payload = payload_class(**json.loads(item["payload"]))
-                event = event_class(
-                    aggregate_id=item["aggregate_id"],
-                    aggregate_type=item["aggregate_type"],
-                    aggregate_version=int(item["aggregate_version"]),
-                    event_id=UUID(item["event_id"]),
-                    event_version=int(item["event_version"]),
-                    payload=payload,
-                )
-                events.append(event)
+            for item in response["Items"]:
+                event_type = item["event_type"]
+                if event_type in EVENT_CLASSES:
+                    event_class, payload_class = EVENT_CLASSES[event_type]
+                    payload = payload_class(**json.loads(item["payload"]))
+                    event = event_class(
+                        aggregate_id=item["aggregate_id"],
+                        aggregate_type=item["aggregate_type"],
+                        aggregate_version=int(item["aggregate_version"]),
+                        event_id=UUID(item["event_id"]),
+                        event_version=int(item["event_version"]),
+                        payload=payload,
+                    )
+                    events.append(event)
 
-        events.sort(key=self._sort_events)
+            events.sort(key=self._sort_events)
 
-        return events
+            return events
+        except Exception as err:
+            return err

src/drivers/timestream_metric_sink.py

@@ -18,7 +18,6 @@ def __init__(self) -> None:
         self.timestream_client = boto3.client("timestream-write")
 
     def store_metrics(self, metrics: List[Metric]) -> Optional[Exception]:
-        logger.msg(f"Writing records for metrics {metrics}")
         current_time = round(time.time() * 1000)
         records = []
         for metric in metrics:
@@ -48,9 +47,5 @@ def store_metrics(self, metrics: List[Metric]) -> Optional[Exception]:
                     Records=records,
                     CommonAttributes={},
                 )
-                logger.msg(
-                    f"WriteRecords Status: {result['ResponseMetadata']['HTTPStatusCode']}"
-                )
         except Exception as err:
-            logger.error(f"Error: {err}")
             return err

src/entities/accounts.py

@@ -13,7 +13,7 @@ class AccountRequested(BaseEvent):
     aggregate_type = "Account"
     event_version = 1
     payload: AccountRequestedPayload
-    event_type = "account_requested"
+    event_type: Literal["account_requested"] = "account_requested"
 
 
 class AccountCreatedPayload(BaseModel):

src/entities/patch.py

@@ -7,6 +7,7 @@
 
 # Events
 class PatchRunSummaryPayload(BaseModel):
+    timestamp: float
     failed_instances: str
     successful_instances: str
 

src/entrypoints/aws_lambda.py

@@ -12,15 +12,24 @@
 
 
 def lambda_handler(event, _) -> None:
-    print(f"event is {event}")
+    logger.msg("Incoming event", incoming_event=event)
     full_event = event["detail"]
-    events = event_source.get_events_for_aggregate(full_event["aggregate_id"])
-    logger.msg("Returned aggregate events", events=events)
-    response = handle_event(full_event, events)
+    aggregate_events = event_source.get_events_for_aggregate(full_event["aggregate_id"])
+    if isinstance(aggregate_events, Exception):
+            logger.error("Exception retrieving aggregate events", exception=str(aggregate_events))
+    logger.msg("Aggregate event/s retrieved", aggregate_count=len(aggregate_events))
+    response = handle_event(full_event, aggregate_events)
     if isinstance(response, Exception):
-        logger.msg("Exception handling event", exception=str(response))
+        logger.error("Exception handling event", exception=str(response))
     else:
         event, metrics = response
-        logger.msg("Received from handling event", response=str(response))
-        event_sink.store_events([event])
-        metric_sink.store_metrics(metrics)
+        logger.msg("Received from handling event", outgoing_event=str(event), outgoing_metrics=str(metrics))
+        response = event_sink.store_events([event])
+        if isinstance(response, Exception):
+            logger.error("Exception storing event", exception=str(response))
+        logger.info("Event stored in DynamoDB", stored_event=str(event))
+        response = metric_sink.store_metrics(metrics)
+        if isinstance(response, Exception):
+            logger.error("Exception storing metric", exception=str(response))
+        logger.info("Metrics stored in Timestream", stored_metrics=str(metrics))
+

src/usecases/accounts.py

@@ -1,41 +1,16 @@
-from typing import List, Dict, Tuple, Union
-from uuid import uuid4
+from typing import List, Tuple
 
 from src.entities.accounts import (
     AccountCreated,
-    AccountCreatedPayload,
     AccountLeadTime,
     AccountRequested,
-    AccountRequestedPayload,
 )
 from src.entities.events import Event
 
 
-def _convert_payload_to_event(
-    event: Dict, aggregate_version: int
-) -> Union[AccountCreated, AccountRequested]:
-    if event["event_type"] == "account_requested":
-        return AccountRequested(
-            aggregate_id=event["aggregate_id"],
-            aggregate_version=aggregate_version + 1,
-            event_id=str(uuid4()),
-            event_version=1,
-            payload=AccountRequestedPayload(timestamp=int(event["time"])),
-        )
-    elif event["event_type"] == "account_created":
-        return AccountCreated(
-            aggregate_id=event["aggregate_id"],
-            aggregate_version=aggregate_version + 1,
-            event_id=str(uuid4()),
-            event_version=1,
-            payload=AccountCreatedPayload(timestamp=int(event["time"])),
-        )
-
-
 def handle_account_created(
-    event: Dict, aggregate_events: List[Event]
+    event: Event, aggregate_events: List[Event]
 ) -> Tuple[AccountCreated, List[AccountLeadTime]]:
-    event = _convert_payload_to_event(event, len(aggregate_events))
     if aggregate_events:
         return (
             event,
@@ -52,7 +27,6 @@ def handle_account_created(
 
 
 def handle_account_requested(
-    event: Dict, aggregate_events: List[Event]
+    event: Event, aggregate_events: List[Event]
 ) -> Tuple[AccountRequested, List]:
-    event = _convert_payload_to_event(event, len(aggregate_events))
     return (event, [])