feat: experimentation flagsmith warehouse setup api scaffolding (#7542)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: Zaimwa9

api/app/settings/common.py

@@ -167,6 +167,7 @@
     "softdelete",
     "metadata",
     "app_analytics",
+    "experimentation",
     "oauth2_metadata",
 ]
 

api/audit/related_object_type.py

@@ -12,3 +12,4 @@ class RelatedObjectType(enum.Enum):
     EF_VERSION = "Environment feature version"
     FEATURE_HEALTH = "Feature health status"
     RELEASE_PIPELINE = "Release pipeline"
+    WAREHOUSE_CONNECTION = "Warehouse connection"

api/environments/urls.py

@@ -173,4 +173,8 @@
         get_experiment_results,
         name="experiment-results",
     ),
+    path(
+        "<str:environment_api_key>/warehouse-connections/",
+        include("experimentation.urls"),
+    ),
 ]

api/experimentation/__init__.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class ExperimentationConfig(AppConfig):
+    name = "experimentation"

api/experimentation/apps.py

@@ -0,0 +1 @@
+WAREHOUSE_CONNECTION_FLAG = "experimentation_warehouse_connection"

api/experimentation/constants.py

@@ -0,0 +1,89 @@
+# Generated by Django 5.2.14 on 2026-05-19 11:05
+
+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):
+
+    initial = True
+
+    dependencies = [
+        ("environments", "0037_add_uuid_field"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="WarehouseConnection",
+            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),
+                ),
+                (
+                    "warehouse_type",
+                    models.CharField(
+                        choices=[
+                            ("flagsmith", "Flagsmith"),
+                            ("snowflake", "Snowflake"),
+                            ("clickhouse", "ClickHouse"),
+                        ],
+                        max_length=50,
+                    ),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("pending_connection", "Pending Connection"),
+                            ("connected", "Connected"),
+                            ("errored", "Errored"),
+                        ],
+                        default="pending_connection",
+                        max_length=50,
+                    ),
+                ),
+                ("name", models.CharField(max_length=255)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "environment",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="warehouse_connections",
+                        to="environments.environment",
+                    ),
+                ),
+            ],
+            options={
+                "constraints": [
+                    models.UniqueConstraint(
+                        condition=models.Q(("deleted_at__isnull", True)),
+                        fields=("warehouse_type", "environment"),
+                        name="unique_active_warehouse_per_type_and_env",
+                    )
+                ],
+            },
+            bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model),
+        ),
+    ]

api/experimentation/migrations/0001_introduce_warehouse_connection.py

@@ -0,0 +1,45 @@
+from django.db import models
+from django_lifecycle import LifecycleModelMixin  # type: ignore[import-untyped]
+
+from core.models import SoftDeleteExportableModel
+from environments.models import Environment
+
+
+class WarehouseType(models.TextChoices):
+    FLAGSMITH = "flagsmith", "Flagsmith"
+    SNOWFLAKE = "snowflake", "Snowflake"
+    CLICKHOUSE = "clickhouse", "ClickHouse"
+
+
+class WarehouseConnectionStatus(models.TextChoices):
+    PENDING_CONNECTION = "pending_connection", "Pending Connection"
+    CONNECTED = "connected", "Connected"
+    ERRORED = "errored", "Errored"
+
+
+class WarehouseConnection(LifecycleModelMixin, SoftDeleteExportableModel):  # type: ignore[misc]
+    environment = models.ForeignKey(
+        Environment,
+        on_delete=models.CASCADE,
+        related_name="warehouse_connections",
+    )
+    warehouse_type = models.CharField(
+        max_length=50,
+        choices=WarehouseType.choices,
+    )
+    status = models.CharField(
+        max_length=50,
+        choices=WarehouseConnectionStatus.choices,
+        default=WarehouseConnectionStatus.PENDING_CONNECTION,
+    )
+    name = models.CharField(max_length=255)
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        constraints = [
+            models.UniqueConstraint(
+                fields=["warehouse_type", "environment"],
+                condition=models.Q(deleted_at__isnull=True),
+                name="unique_active_warehouse_per_type_and_env",
+            ),
+        ]

api/experimentation/migrations/__init__.py

@@ -0,0 +1,23 @@
+from rest_framework.permissions import BasePermission
+from rest_framework.request import Request
+from rest_framework.views import APIView
+
+from environments.models import Environment
+from experimentation.services import is_warehouse_feature_enabled
+from users.models import FFAdminUser
+
+
+class WarehouseConnectionPermission(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_warehouse_feature_enabled(environment.project.organisation):
+            return False
+
+        user: FFAdminUser = request.user  # type: ignore[assignment]
+        return user.is_environment_admin(environment)