ref(crons): Make SimpleCheckIn a dataclass (#80741)

Turning this into a dataclass since we'll be serializing and
deserializing into this datatype for incident occurrences

src/sentry/monitors/logic/incident_occurrence.py

@@ -137,9 +137,9 @@ def get_failure_reason(failed_checkins: Sequence[SimpleCheckIn]):
     """
 
     status_counts = Counter(
-        checkin["status"]
+        checkin.status
         for checkin in failed_checkins
-        if checkin["status"] in HUMAN_FAILURE_STATUS_MAP.keys()
+        if checkin.status in HUMAN_FAILURE_STATUS_MAP.keys()
     )
 
     if sum(status_counts.values()) == 1:

src/sentry/monitors/logic/incidents.py

@@ -2,7 +2,6 @@
 
 import logging
 from datetime import datetime
-from typing import cast
 
 from sentry.monitors.logic.incident_occurrence import create_incident_occurrence
 from sentry.monitors.models import CheckInStatus, MonitorCheckIn, MonitorIncident, MonitorStatus
@@ -29,30 +28,25 @@ def try_incident_threshold(
     # check to see if we need to update the status
     if monitor_env.status in [MonitorStatus.OK, MonitorStatus.ACTIVE]:
         if failure_issue_threshold == 1:
-            previous_checkins: list[SimpleCheckIn] = [
-                {
-                    "id": failed_checkin.id,
-                    "date_added": failed_checkin.date_added,
-                    "status": failed_checkin.status,
-                }
-            ]
+            previous_checkins: list[SimpleCheckIn] = [failed_checkin.as_simple_checkin()]
         else:
-            previous_checkins = cast(
-                list[SimpleCheckIn],
+            previous_checkins = [
+                SimpleCheckIn(**row)
+                for row in
                 # Using .values for performance reasons
                 MonitorCheckIn.objects.filter(
                     monitor_environment=monitor_env, date_added__lte=failed_checkin.date_added
                 )
                 .order_by("-date_added")
-                .values("id", "date_added", "status"),
-            )
+                .values("id", "date_added", "status")
+            ]
 
             # reverse the list after slicing in order to start with oldest check-in
             previous_checkins = list(reversed(previous_checkins[:failure_issue_threshold]))
 
             # If we have any successful check-ins within the threshold of
             # commits we have NOT reached an incident state
-            if any([checkin["status"] == CheckInStatus.OK for checkin in previous_checkins]):
+            if any([checkin.status == CheckInStatus.OK for checkin in previous_checkins]):
                 return False
 
         # change monitor status + update fingerprint timestamp
@@ -67,21 +61,15 @@ def try_incident_threshold(
             resolving_checkin=None,
             defaults={
                 "monitor": monitor_env.monitor,
-                "starting_checkin_id": starting_checkin["id"],
-                "starting_timestamp": starting_checkin["date_added"],
+                "starting_checkin_id": starting_checkin.id,
+                "starting_timestamp": starting_checkin.date_added,
             },
         )
 
     elif monitor_env.status == MonitorStatus.ERROR:
         # if monitor environment has a failed status, use the failed
         # check-in and send occurrence
-        previous_checkins = [
-            {
-                "id": failed_checkin.id,
-                "date_added": failed_checkin.date_added,
-                "status": failed_checkin.status,
-            }
-        ]
+        previous_checkins = [failed_checkin.as_simple_checkin()]
 
         # get the active incident from the monitor environment
         incident = monitor_env.active_incident
@@ -93,7 +81,7 @@ def try_incident_threshold(
     # - We have an active incident and fingerprint
     # - The monitor and env are not muted
     if not monitor_env.monitor.is_muted and not monitor_env.is_muted and incident:
-        checkins = MonitorCheckIn.objects.filter(id__in=[c["id"] for c in previous_checkins])
+        checkins = MonitorCheckIn.objects.filter(id__in=[c.id for c in previous_checkins])
         for checkin in checkins:
             create_incident_occurrence(
                 previous_checkins,

src/sentry/monitors/models.py

@@ -37,7 +37,7 @@
 from sentry.locks import locks
 from sentry.models.environment import Environment
 from sentry.models.rule import Rule, RuleSource
-from sentry.monitors.types import CrontabSchedule, IntervalSchedule
+from sentry.monitors.types import CrontabSchedule, IntervalSchedule, SimpleCheckIn
 from sentry.types.actor import Actor
 from sentry.utils.retries import TimedRetryPolicy
 
@@ -581,6 +581,9 @@ def save(self, *args, **kwargs):
     def _update_timestamps(self):
         pass
 
+    def as_simple_checkin(self) -> SimpleCheckIn:
+        return SimpleCheckIn(self.id, self.date_added, self.status)
+
 
 def delete_file_for_monitorcheckin(instance: MonitorCheckIn, **kwargs):
     if file_id := instance.attachment_id:

src/sentry/monitors/types.py

@@ -101,7 +101,8 @@ def from_dict(cls, data: CheckinItemData) -> CheckinItem:
         )
 
 
-class SimpleCheckIn(TypedDict):
+@dataclass
+class SimpleCheckIn:
     """
     A stripped down check in object
     """