feat(aci): Add ability to create new environments in the API

src/sentry/incidents/serializers/alert_rule.py

@@ -1,6 +1,12 @@
+from __future__ import annotations
+
 import logging
 import operator
 from datetime import timedelta
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from sentry.models.environment import Environment
 
 import sentry_sdk
 from django import forms
@@ -54,7 +60,7 @@ class AlertRuleSerializer(SnubaQueryValidator, CamelSnakeModelSerializer[AlertRu
      - `user`: The user from `request.user`
     """
 
-    environment = EnvironmentField(required=False, allow_null=True)
+    environment = serializers.CharField(required=False, allow_null=True)
     projects = serializers.ListField(
         child=ProjectField(scope="project:read"),
         required=False,
@@ -120,6 +126,11 @@ class Meta:
         AlertRuleThresholdType.BELOW: lambda threshold: 100 - threshold,
     }
 
+    def validate_environment(self, value: str | None) -> Environment | None:
+        field = EnvironmentField()
+        field.bind("environment", self)
+        return field.to_internal_value(value)
+
     def validate_threshold_type(self, threshold_type):
         try:
             return AlertRuleThresholdType(threshold_type)

src/sentry/snuba/snuba_query_validator.py

@@ -9,7 +9,6 @@
 from snuba_sdk import Column, Condition, Entity, Limit, Op
 
 from sentry import features
-from sentry.api.serializers.rest_framework import EnvironmentField
 from sentry.exceptions import (
     IncompatibleMetricsQuery,
     InvalidSearchQuery,
@@ -23,6 +22,7 @@
     translate_aggregate_field,
 )
 from sentry.incidents.utils.constants import INCIDENTS_SNUBA_SUBSCRIPTION_TYPE
+from sentry.models.environment import Environment
 from sentry.models.project import Project
 from sentry.search.eap.constants import VALID_GRANULARITIES
 from sentry.search.eap.trace_metrics.validator import validate_trace_metrics_aggregate
@@ -90,7 +90,11 @@
     query = serializers.CharField(required=True, allow_blank=True)
     aggregate = serializers.CharField(required=True)
     time_window = serializers.IntegerField(required=True)
-    environment = EnvironmentField(required=True, allow_null=True)
+    environment = serializers.CharField(
+        required=True,
+        allow_null=True,
+        max_length=64,
+    )
     event_types = serializers.ListField(
         child=serializers.CharField(),
     )
@@ -124,6 +128,23 @@
         # TODO: only accept time_window in seconds once AlertRuleSerializer is removed
         self.time_window_seconds = timeWindowSeconds
 
+    def validate_environment(self, value: str | None) -> Environment | None:
+        """
+        This is not using the `EnvironmentField` so we can inline create new envs
+        inline when creating alerts. The use case is when a new environment is needed
+        for an alert, but there haven't been any events ingested yet.
+        """
+        if value is None:
+            return None
+
+        try:
+            return Environment.get_or_create(
+                project=self.context["project"],
+                name=value,
+            )
+        except Exception:
+            raise serializers.ValidationError("Failed to retrieve or create environment.")
+
     def validate_aggregate(self, aggregate: str) -> str:
         """
         Reject upsampled_count() as user input. This function is reserved for internal use

tests/sentry/snuba/test_validators.py

@@ -2,6 +2,7 @@
 from rest_framework import serializers
 from rest_framework.exceptions import ErrorDetail
 
+from sentry.models.environment import Environment
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import SnubaQuery, SnubaQueryEventType
 from sentry.snuba.snuba_query_validator import SnubaQueryValidator
@@ -39,6 +40,21 @@ def test_simple(self) -> None:
         assert validator.validated_data["event_types"] == [SnubaQueryEventType.EventType.ERROR]
         assert isinstance(validator.validated_data["_creator"], DataSourceCreator)
 
+    def test_environment_get_or_create(self) -> None:
+        new_env_name = "new-test-environment"
+        assert not Environment.objects.filter(
+            name=new_env_name, organization_id=self.project.organization_id
+        ).exists()
+
+        self.valid_data["environment"] = new_env_name
+        validator = SnubaQueryValidator(data=self.valid_data, context=self.context)
+        assert validator.is_valid()
+
+        env = validator.validated_data["environment"]
+        assert isinstance(env, Environment)
+        assert env.name == new_env_name
+        assert env.organization_id == self.project.organization_id
+
     def test_invalid_query(self) -> None:
         unsupported_query = "release:latest"
         self.valid_data["query"] = unsupported_query