feat(experimentation): add Experiment base model and CRUD endpoints (#7591)

Author: Zaimwa9

api/audit/related_object_type.py

@@ -13,3 +13,4 @@ class RelatedObjectType(enum.Enum):
     FEATURE_HEALTH = "Feature health status"
     RELEASE_PIPELINE = "Release pipeline"
     WAREHOUSE_CONNECTION = "Warehouse connection"
+    EXPERIMENT = "Experiment"

api/environments/urls.py

@@ -177,4 +177,8 @@
         "<str:environment_api_key>/warehouse-connections/",
         include("experimentation.urls"),
     ),
+    path(
+        "<str:environment_api_key>/experiments/",
+        include("experimentation.experiment_urls"),
+    ),
 ]

api/experimentation/constants.py

@@ -1 +1,2 @@
 WAREHOUSE_CONNECTION_FLAG = "experimentation_warehouse_connection"
+EXPERIMENT_FLAG = "experimental_flags"

api/experimentation/experiment_urls.py

@@ -0,0 +1,10 @@
+from rest_framework.routers import DefaultRouter
+
+from experimentation.views import ExperimentViewSet
+
+app_name = "experiments"
+
+router = DefaultRouter()
+router.register(r"", ExperimentViewSet, basename="experiments")
+
+urlpatterns = router.urls

api/experimentation/migrations/0004_experiment.py

@@ -0,0 +1,94 @@
+# Generated by Django 5.2.14 on 2026-05-25 08:45
+
+import django.db.models.deletion
+import django_lifecycle.mixins  # type: ignore[import-untyped]
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("environments", "0037_add_uuid_field"),
+        ("experimentation", "0003_unique_connection_per_environment"),
+        ("features", "0066_constrain_feature_type"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Experiment",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "deleted_at",
+                    models.DateTimeField(
+                        blank=True,
+                        db_index=True,
+                        default=None,
+                        editable=False,
+                        null=True,
+                    ),
+                ),
+                (
+                    "uuid",
+                    models.UUIDField(default=uuid.uuid4, editable=False, unique=True),
+                ),
+                ("name", models.CharField(max_length=255)),
+                ("hypothesis", models.TextField()),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("created", "Created"),
+                            ("running", "Running"),
+                            ("paused", "Paused"),
+                            ("completed", "Completed"),
+                        ],
+                        default="created",
+                        max_length=50,
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("started_at", models.DateTimeField(blank=True, null=True)),
+                ("ended_at", models.DateTimeField(blank=True, null=True)),
+                (
+                    "environment",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="experiments",
+                        to="environments.environment",
+                    ),
+                ),
+                (
+                    "feature",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="experiments",
+                        to="features.feature",
+                    ),
+                ),
+            ],
+            options={
+                "constraints": [
+                    models.UniqueConstraint(
+                        condition=models.Q(
+                            ("deleted_at__isnull", True),
+                            models.Q(("status", "completed"), _negated=True),
+                        ),
+                        fields=("feature", "environment"),
+                        name="unique_active_experiment_per_feature_env",
+                    )
+                ],
+            },
+            bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model),
+        ),
+    ]

api/experimentation/models.py

@@ -1,4 +1,5 @@
 from django.db import models
+from django.db.models import Q
 from django_lifecycle import (  # type: ignore[import-untyped]
     AFTER_CREATE,
     AFTER_DELETE,
@@ -68,3 +69,51 @@ def sync_to_ingestion_on_delete(self) -> None:
         delete_environment_key_from_ingestion.delay(
             kwargs={"environment_api_key": self.environment.api_key},
         )
+
+
+class ExperimentStatus(models.TextChoices):
+    CREATED = "created", "Created"
+    RUNNING = "running", "Running"
+    PAUSED = "paused", "Paused"
+    COMPLETED = "completed", "Completed"
+
+
+VALID_STATUS_TRANSITIONS: dict[str, set[str]] = {
+    ExperimentStatus.CREATED: {ExperimentStatus.RUNNING},
+    ExperimentStatus.RUNNING: {ExperimentStatus.PAUSED, ExperimentStatus.COMPLETED},
+    ExperimentStatus.PAUSED: {ExperimentStatus.RUNNING, ExperimentStatus.COMPLETED},
+    ExperimentStatus.COMPLETED: set(),
+}
+
+
+class Experiment(LifecycleModelMixin, SoftDeleteExportableModel):  # type: ignore[misc]
+    environment = models.ForeignKey(
+        Environment,
+        on_delete=models.CASCADE,
+        related_name="experiments",
+    )
+    feature = models.ForeignKey(
+        "features.Feature",
+        on_delete=models.CASCADE,
+        related_name="experiments",
+    )
+    name = models.CharField(max_length=255)
+    hypothesis = models.TextField()
+    status = models.CharField(
+        max_length=50,
+        choices=ExperimentStatus.choices,
+        default=ExperimentStatus.CREATED,
+    )
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+    started_at = models.DateTimeField(null=True, blank=True)
+    ended_at = models.DateTimeField(null=True, blank=True)
+
+    class Meta:
+        constraints = [
+            models.UniqueConstraint(
+                fields=["feature", "environment"],
+                condition=Q(deleted_at__isnull=True) & ~Q(status="completed"),
+                name="unique_active_experiment_per_feature_env",
+            ),
+        ]

api/experimentation/permissions.py

@@ -3,7 +3,10 @@
 from rest_framework.views import APIView
 
 from environments.models import Environment
-from experimentation.services import is_warehouse_feature_enabled
+from experimentation.services import (
+    is_experiment_feature_enabled,
+    is_warehouse_feature_enabled,
+)
 from users.models import FFAdminUser
 
 
@@ -21,3 +24,19 @@ def has_permission(self, request: Request, view: APIView) -> bool:
 
         user: FFAdminUser = request.user  # type: ignore[assignment]
         return user.is_environment_admin(environment)
+
+
+class ExperimentPermission(BasePermission):
+    def has_permission(self, request: Request, view: APIView) -> bool:
+        try:
+            environment = Environment.objects.get(
+                api_key=view.kwargs.get("environment_api_key")
+            )
+        except Environment.DoesNotExist:
+            return False
+
+        if not is_experiment_feature_enabled(environment.project.organisation):
+            return False
+
+        user: FFAdminUser = request.user  # type: ignore[assignment]
+        return user.is_environment_admin(environment)

api/experimentation/serializers.py

@@ -4,10 +4,12 @@
 
 from environments.models import Environment
 from experimentation.models import (
+    Experiment,
     WarehouseConnection,
     WarehouseType,
 )
 from experimentation.types import SNOWFLAKE_DEFAULTS, SnowflakeConfig
+from features.feature_types import MULTIVARIATE
 
 
 class WarehouseConnectionSerializer(serializers.ModelSerializer):  # type: ignore[type-arg]
@@ -70,3 +72,46 @@ def _validate_snowflake_config(config: dict[str, Any]) -> SnowflakeConfig:
             **config,  # type: ignore[typeddict-item]
         }
         return merged
+
+
+class ExperimentSerializer(serializers.ModelSerializer):  # type: ignore[type-arg]
+    class Meta:
+        model = Experiment
+        fields = (
+            "id",
+            "feature",
+            "name",
+            "hypothesis",
+            "status",
+            "created_at",
+            "updated_at",
+            "started_at",
+            "ended_at",
+        )
+        read_only_fields = (
+            "id",
+            "status",
+            "created_at",
+            "updated_at",
+            "started_at",
+            "ended_at",
+        )
+
+    def validate_feature(self, feature: Any) -> Any:
+        if feature.type != MULTIVARIATE:
+            raise serializers.ValidationError(
+                "Experiments can only be created for multivariate flags."
+            )
+        environment: Environment | None = self.context.get("environment")
+        if environment and feature.project_id != environment.project_id:
+            raise serializers.ValidationError(
+                "Feature does not belong to this environment's project."
+            )
+        return feature
+
+    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
+        if self.instance is not None and "feature" in attrs:
+            raise serializers.ValidationError(
+                {"feature": "Cannot change the feature of an existing experiment."}
+            )
+        return attrs

api/experimentation/services.py

@@ -2,13 +2,15 @@
 
 import typing
 
+from django.utils import timezone
+
 from audit.models import AuditLog
 from audit.related_object_type import RelatedObjectType
-from experimentation.constants import WAREHOUSE_CONNECTION_FLAG
+from experimentation.constants import EXPERIMENT_FLAG, WAREHOUSE_CONNECTION_FLAG
 from integrations.flagsmith.client import get_openfeature_client
 
 if typing.TYPE_CHECKING:
-    from experimentation.models import WarehouseConnection
+    from experimentation.models import Experiment, WarehouseConnection
     from organisations.models import Organisation
     from users.models import FFAdminUser
 
@@ -21,6 +23,22 @@ def is_warehouse_feature_enabled(organisation: Organisation) -> bool:
     )
 
 
+def is_experiment_feature_enabled(organisation: Organisation) -> bool:
+    return get_openfeature_client().get_boolean_value(
+        EXPERIMENT_FLAG,
+        default_value=False,
+        evaluation_context=organisation.openfeature_evaluation_context,
+    )
+
+
+def _resolve_audit_log_author(
+    user: FFAdminUser,
+) -> dict[str, int | None]:
+    if getattr(user, "is_master_api_key_user", False):
+        return {"author_id": None, "master_api_key_id": user.key.id}
+    return {"author_id": user.pk, "master_api_key_id": None}
+
+
 def create_warehouse_audit_log(
     connection: WarehouseConnection,
     user: FFAdminUser,
@@ -30,11 +48,55 @@ def create_warehouse_audit_log(
     AuditLog.objects.create(
         environment=connection.environment,
         project=connection.environment.project,
-        author=user,
+        **_resolve_audit_log_author(user),
         related_object_id=connection.id,
         related_object_type=RelatedObjectType.WAREHOUSE_CONNECTION.name,
         log=(
             f"Warehouse connection {action} for environment "
             f"{connection.environment.name}"
         ),
     )
+
+
+def create_experiment_audit_log(
+    experiment: Experiment,
+    user: FFAdminUser,
+    *,
+    action: str,
+) -> None:
+    AuditLog.objects.create(
+        environment=experiment.environment,
+        project=experiment.environment.project,
+        **_resolve_audit_log_author(user),
+        related_object_id=experiment.id,
+        related_object_type=RelatedObjectType.EXPERIMENT.name,
+        log=(
+            f"Experiment '{experiment.name}' {action} for environment "
+            f"{experiment.environment.name}"
+        ),
+    )
+
+
+def transition_experiment_status(
+    experiment: Experiment,
+    target_status: str,
+    user: FFAdminUser,
+) -> Experiment:
+    from experimentation.models import VALID_STATUS_TRANSITIONS, ExperimentStatus
+
+    valid_targets = VALID_STATUS_TRANSITIONS.get(experiment.status, set())
+    if target_status not in valid_targets:
+        raise ValueError(
+            f"Cannot transition from '{experiment.status}' to '{target_status}'."
+        )
+
+    experiment.status = target_status
+
+    if target_status == ExperimentStatus.RUNNING and not experiment.started_at:
+        experiment.started_at = timezone.now()
+    elif target_status == ExperimentStatus.COMPLETED:
+        experiment.ended_at = timezone.now()
+
+    experiment.save()
+    create_experiment_audit_log(experiment, user, action=target_status)
+    return experiment