diff --git a/src/sentry/workflow_engine/models/data_source.py b/src/sentry/workflow_engine/models/data_source.py
index 320a01709a290e..f9dcea141d2344 100644
--- a/src/sentry/workflow_engine/models/data_source.py
+++ b/src/sentry/workflow_engine/models/data_source.py
@@ -3,6 +3,8 @@
 from typing import Generic, TypeVar
 
 from django.db import models
+from django.db.models.signals import pre_save
+from django.dispatch import receiver
 
 from sentry.backup.scopes import RelocationScope
 from sentry.db.models import (
@@ -11,6 +13,7 @@
     FlexibleForeignKey,
     region_silo_model,
 )
+from sentry.utils.registry import NoRegistrationExistsError
 from sentry.workflow_engine.models.data_source_detector import DataSourceDetector
 from sentry.workflow_engine.registry import data_source_type_registry
 from sentry.workflow_engine.types import DataSourceTypeHandler
@@ -33,7 +36,7 @@ class DataSource(DefaultFieldsModel):
     # Should this be a string so we can support UUID / ints?
     query_id = BoundedBigIntegerField()
 
-    # TODO - Add a type here
+    # This is a dynamic field, depending on the type in the data_source_type_registry
     type = models.TextField()
 
     detectors = models.ManyToManyField("workflow_engine.Detector", through=DataSourceDetector)
@@ -49,3 +52,19 @@ def type_handler(self) -> builtins.type[DataSourceTypeHandler]:
         if not handler:
             raise ValueError(f"Unknown data source type: {self.type}")
         return handler
+
+
+@receiver(pre_save, sender=DataSource)
+def ensure_valid_type(sender, instance: DataSource, **kwargs):
+    """
+    Ensure that the type of the data source is valid and registered in the data_source_type_registry
+    """
+    data_source_type = instance.type
+
+    if not data_source_type:
+        raise ValueError(f"No group type found with type {instance.type}")
+
+    try:
+        data_source_type_registry.get(data_source_type)
+    except NoRegistrationExistsError:
+        raise ValueError(f"No data source type found with type {data_source_type}")