use TypedDicts to ensure values are correct

Author: saponifi3d

src/sentry/workflow_engine/models/data_condition.py

@@ -3,7 +3,7 @@
 import time
 from datetime import timedelta
 from enum import StrEnum
-from typing import Any, TypeVar, cast
+from typing import Any, TypedDict, TypeVar, cast
 
 from django.db import models
 from django.db.models.signals import pre_save
@@ -111,6 +111,13 @@ class Condition(StrEnum):
 FAST_CONDITION_TOO_SLOW_THRESHOLD = timedelta(milliseconds=500)
 
 
+class DataConditionSnapshot(TypedDict):
+    id: int
+    type: str
+    comparison: str
+    condition_result: DataConditionResult
+
+
 @region_silo_model
 class DataCondition(DefaultFieldsModel):
     """
@@ -137,7 +144,7 @@ class DataCondition(DefaultFieldsModel):
         on_delete=models.CASCADE,
     )
 
-    def get_snapshot(self) -> dict[str, Any]:
+    def get_snapshot(self) -> DataConditionSnapshot:
         return {
             "id": self.id,
             "type": self.type,

src/sentry/workflow_engine/models/data_condition_group.py

@@ -1,11 +1,20 @@
+from __future__ import annotations
+
 from enum import StrEnum
-from typing import Any, ClassVar, Self
+from typing import ClassVar, Self, TypedDict
 
 from django.db import models
 
 from sentry.backup.scopes import RelocationScope
 from sentry.db.models import DefaultFieldsModel, region_silo_model, sane_repr
 from sentry.db.models.manager.base import BaseManager
+from sentry.workflow_engine.models.data_condition import DataConditionSnapshot
+
+
+class DataConditionGroupSnapshot(TypedDict):
+    id: int
+    logic_type: DataConditionGroup.Type
+    conditions: list[DataConditionSnapshot]
 
 
 @region_silo_model
@@ -37,13 +46,13 @@ class Type(StrEnum):
     )
     organization = models.ForeignKey("sentry.Organization", on_delete=models.CASCADE)
 
-    def get_snapshot(self) -> dict[str, Any]:
+    def get_snapshot(self) -> DataConditionGroupSnapshot:
         conditions = []
         if hasattr(self, "conditions"):
             conditions = [cond.get_snapshot() for cond in self.conditions.all()]
 
         return {
             "id": self.id,
-            "logic_type": self.logic_type,
+            "logic_type": DataConditionGroup.Type(self.logic_type),
             "conditions": conditions,
         }

src/sentry/workflow_engine/models/detector.py

@@ -3,7 +3,7 @@
 import builtins
 import logging
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar, TypedDict
 
 from django.conf import settings
 from django.db import models
@@ -29,10 +29,18 @@
 
 if TYPE_CHECKING:
     from sentry.workflow_engine.handlers.detector import DetectorHandler
+    from sentry.workflow_engine.models.data_condition_group import DataConditionGroupSnapshot
 
 logger = logging.getLogger(__name__)
 
 
+class DetectorSnapshot(TypedDict):
+    id: int
+    enabled: bool
+    status: int
+    trigger_condition: DataConditionGroupSnapshot | None
+
+
 class DetectorManager(BaseManager["Detector"]):
     def get_queryset(self) -> BaseQuerySet[Detector]:
         return (
@@ -141,16 +149,16 @@ def settings(self) -> DetectorSettings:
 
         return settings
 
-    def get_snapshot(self) -> dict[str, Any]:
-        trigger_conditions = None
+    def get_snapshot(self) -> DetectorSnapshot:
+        trigger_condition = None
         if self.workflow_condition_group:
-            trigger_conditions = self.workflow_condition_group.get_snapshot()
+            trigger_condition = self.workflow_condition_group.get_snapshot()
 
         return {
             "id": self.id,
             "enabled": self.enabled,
             "status": self.status,
-            "trigger_conditions": trigger_conditions,
+            "trigger_condition": trigger_condition,
         }
 
     def get_audit_log_data(self) -> dict[str, Any]:

src/sentry/workflow_engine/models/workflow.py

@@ -2,7 +2,7 @@
 
 import logging
 from dataclasses import replace
-from typing import Any, ClassVar
+from typing import Any, ClassVar, TypedDict
 
 from django.conf import settings
 from django.db import models
@@ -17,7 +17,10 @@
 from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.models.owner_base import OwnerModel
 from sentry.workflow_engine.models.data_condition import DataCondition, is_slow_condition
-from sentry.workflow_engine.models.data_condition_group import DataConditionGroup
+from sentry.workflow_engine.models.data_condition_group import (
+    DataConditionGroup,
+    DataConditionGroupSnapshot,
+)
 from sentry.workflow_engine.processors.data_condition_group import TriggerResult
 from sentry.workflow_engine.types import ConditionError, WorkflowEventData
 
@@ -26,6 +29,14 @@
 logger = logging.getLogger(__name__)
 
 
+class WorkflowSnapshot(TypedDict):
+    id: int
+    enabled: bool
+    environment_id: int | None
+    status: int
+    triggers: DataConditionGroupSnapshot | None
+
+
 class WorkflowManager(BaseManager["Workflow"]):
     def get_queryset(self) -> BaseQuerySet[Workflow]:
         return (
@@ -92,7 +103,7 @@ class Meta:
     def get_audit_log_data(self) -> dict[str, Any]:
         return {"name": self.name}
 
-    def get_snapshot(self) -> dict[str, Any]:
+    def get_snapshot(self) -> WorkflowSnapshot:
         when_condition_group = None
         if self.when_condition_group:
             when_condition_group = self.when_condition_group.get_snapshot()

src/sentry/workflow_engine/processors/workflow.py

@@ -117,7 +117,7 @@ def enqueue_workflows(
     for queue_item in items_by_workflow.values():
         if not queue_item.delayed_if_group_ids and not queue_item.passing_if_group_ids:
             # Skip because there are no IF groups we could possibly fire actions for if
-            # the WHEN/IF delayed condtions are met
+            # the WHEN/IF delayed conditions are met
             continue
         project_id = queue_item.event.project_id
         items_by_project_id[project_id].append(queue_item)