feat(crons): Add CronMonitorDataSourceHandler (#97548)

Adding a new `DataSourceHandler` so that we can treat cron `Monitor`
rows as a `DataSource`

<!-- Describe your PR here. -->

Author: wedamija

src/sentry/monitors/models.py

@@ -5,7 +5,7 @@
 import zoneinfo
 from collections.abc import Sequence
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, ClassVar, Self
+from typing import TYPE_CHECKING, Any, ClassVar, Self, override
 from uuid import uuid4
 
 import jsonschema
@@ -34,12 +34,17 @@
 from sentry.db.models.fields.slug import DEFAULT_SLUG_MAX_LENGTH, SentrySlugField
 from sentry.db.models.manager.base import BaseManager
 from sentry.db.models.utils import slugify_instance
+from sentry.deletions.base import ModelRelation
 from sentry.locks import locks
 from sentry.models.environment import Environment
+from sentry.models.organization import Organization
 from sentry.models.rule import Rule, RuleSource
-from sentry.monitors.types import CrontabSchedule, IntervalSchedule
+from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR, CrontabSchedule, IntervalSchedule
 from sentry.types.actor import Actor
 from sentry.utils.retries import TimedRetryPolicy
+from sentry.workflow_engine.models import DataSource
+from sentry.workflow_engine.registry import data_source_type_registry
+from sentry.workflow_engine.types import DataSourceTypeHandler
 
 logger = logging.getLogger(__name__)
 
@@ -790,3 +795,41 @@ class MonitorEnvBrokenDetection(Model):
     class Meta:
         app_label = "monitors"
         db_table = "sentry_monitorenvbrokendetection"
+
+
+@data_source_type_registry.register(DATA_SOURCE_CRON_MONITOR)
+class CronMonitorDataSourceHandler(DataSourceTypeHandler[Monitor]):
+    @staticmethod
+    def bulk_get_query_object(
+        data_sources: list[DataSource],
+    ) -> dict[int, Monitor | None]:
+        monitor_ids: list[int] = []
+
+        for ds in data_sources:
+            try:
+                monitor_ids.append(int(ds.source_id))
+            except ValueError:
+                logger.exception(
+                    "Invalid DataSource.source_id fetching Monitor",
+                    extra={"id": ds.id, "source_id": ds.source_id},
+                )
+
+        qs_lookup = {
+            str(monitor.id): monitor for monitor in Monitor.objects.filter(id__in=monitor_ids)
+        }
+        return {ds.id: qs_lookup.get(ds.source_id) for ds in data_sources}
+
+    @staticmethod
+    def related_model(instance) -> list[ModelRelation]:
+        return [ModelRelation(Monitor, {"id": instance.source_id})]
+
+    @override
+    @staticmethod
+    def get_instance_limit(org: Organization) -> int | None:
+        return None
+
+    @override
+    @staticmethod
+    def get_current_instance_count(org: Organization) -> int:
+        # We don't have a limit at the moment, so no need to count.
+        raise NotImplementedError

src/sentry/monitors/types.py

@@ -10,6 +10,8 @@
 
 from sentry.db.models.fields.slug import DEFAULT_SLUG_MAX_LENGTH
 
+DATA_SOURCE_CRON_MONITOR = "cron_monitor"
+
 
 class CheckinTrace(TypedDict):
     trace_id: str

tests/sentry/monitors/test_models.py

@@ -1,18 +1,22 @@
 from datetime import datetime, timezone
+from unittest import mock
 
 import pytest
 from django.conf import settings
 from django.test.utils import override_settings
 
 from sentry.monitors.models import (
+    CronMonitorDataSourceHandler,
     Monitor,
     MonitorEnvironment,
     MonitorEnvironmentLimitsExceeded,
     MonitorLimitsExceeded,
     ScheduleType,
 )
+from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
 from sentry.monitors.validators import ConfigValidator
 from sentry.testutils.cases import TestCase
+from sentry.workflow_engine.models import DataSource
 
 
 class MonitorTestCase(TestCase):
@@ -270,3 +274,93 @@ def test_config_validator(self) -> None:
         validated_config["bad_key"] = 100
         monitor.config = validated_config
         assert monitor.get_validated_config() is None
+
+
+class CronMonitorDataSourceHandlerTest(TestCase):
+    def setUp(self):
+        super().setUp()
+        self.monitor = self.create_monitor(
+            project=self.project,
+            name="Test Monitor",
+        )
+
+        self.data_source = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            source_id=str(self.monitor.id),
+            organization_id=self.organization.id,
+        )
+
+    def test_bulk_get_query_object(self):
+        result = CronMonitorDataSourceHandler.bulk_get_query_object([self.data_source])
+        assert result[self.data_source.id] == self.monitor
+
+    def test_bulk_get_query_object__multiple_monitors(self):
+        monitor2 = self.create_monitor(
+            project=self.project,
+            name="Test Monitor 2",
+        )
+        data_source2 = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            source_id=str(monitor2.id),
+            organization_id=self.organization.id,
+        )
+
+        data_sources = [self.data_source, data_source2]
+        result = CronMonitorDataSourceHandler.bulk_get_query_object(data_sources)
+
+        assert result[self.data_source.id] == self.monitor
+        assert result[data_source2.id] == monitor2
+
+    def test_bulk_get_query_object__incorrect_data_source(self):
+        ds_with_invalid_monitor_id = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            source_id="not_an_int",
+            organization_id=self.organization.id,
+        )
+
+        with mock.patch("sentry.monitors.models.logger.exception") as mock_logger:
+            data_sources = [self.data_source, ds_with_invalid_monitor_id]
+            result = CronMonitorDataSourceHandler.bulk_get_query_object(data_sources)
+
+            assert result[self.data_source.id] == self.monitor
+            assert result[ds_with_invalid_monitor_id.id] is None
+
+            mock_logger.assert_called_once_with(
+                "Invalid DataSource.source_id fetching Monitor",
+                extra={
+                    "id": ds_with_invalid_monitor_id.id,
+                    "source_id": ds_with_invalid_monitor_id.source_id,
+                },
+            )
+
+    def test_bulk_get_query_object__missing_monitor(self):
+        ds_with_deleted_monitor = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            source_id="99999999",
+            organization_id=self.organization.id,
+        )
+
+        data_sources = [self.data_source, ds_with_deleted_monitor]
+        result = CronMonitorDataSourceHandler.bulk_get_query_object(data_sources)
+
+        assert result[self.data_source.id] == self.monitor
+        assert result[ds_with_deleted_monitor.id] is None
+
+    def test_bulk_get_query_object__empty_list(self):
+        result = CronMonitorDataSourceHandler.bulk_get_query_object([])
+        assert result == {}
+
+    def test_related_model(self):
+        relations = CronMonitorDataSourceHandler.related_model(self.data_source)
+        assert len(relations) == 1
+        relation = relations[0]
+
+        assert relation.params["model"] == Monitor
+        assert relation.params["query"] == {"id": self.data_source.source_id}
+
+    def test_get_instance_limit(self):
+        assert CronMonitorDataSourceHandler.get_instance_limit(self.organization) is None
+
+    def test_get_current_instance_count(self):
+        with pytest.raises(NotImplementedError):
+            CronMonitorDataSourceHandler.get_current_instance_count(self.organization)