feat: ClickHouse bring-your-own-warehouse connections

api/app/settings/common.py

@@ -361,6 +361,7 @@
         "invite": "10/min",
         "user": USER_THROTTLE_RATE,
         "influx_query": "5/min",
+        "warehouse_connection_test": "10/min",
     },
     "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
     "DEFAULT_RENDERER_CLASSES": [

api/app/settings/test.py

@@ -27,6 +27,7 @@
     "user": "100000/day",
     "master_api_key": "100000/day",
     "influx_query": "50/min",
+    "warehouse_connection_test": "100/min",
 }
 
 AWS_SSE_LOGS_BUCKET_NAME = "test_bucket"

api/core/fields.py

@@ -0,0 +1,45 @@
+import base64
+import hashlib
+import json
+from typing import Any
+
+import structlog
+from cryptography.fernet import Fernet, InvalidToken
+from django.conf import settings
+from django.db import models
+
+logger = structlog.get_logger("core")
+
+
+def _get_fernet() -> Fernet:
+    digest = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
+    return Fernet(base64.urlsafe_b64encode(digest))
+
+
+class EncryptedJSONField(models.TextField[Any, Any]):
+    def get_prep_value(self, value: Any) -> str | None:
+        if value is None:
+            return None
+        return _get_fernet().encrypt(json.dumps(value).encode()).decode()
+
+    def from_db_value(
+        self,
+        value: str | None,
+        expression: object,
+        connection: object,
+    ) -> Any:
+        if value is None:
+            return None
+        try:
+            plaintext = _get_fernet().decrypt(value.encode())
+        except InvalidToken:
+            logger.warning("encrypted_field.decrypt_failed", exc_info=True)
+            return None
+        return json.loads(plaintext)
+
+    def get_lookup(self, lookup_name: str) -> Any:
+        if lookup_name != "isnull":
+            raise NotImplementedError(
+                "EncryptedJSONField only supports isnull lookups."
+            )
+        return super().get_lookup(lookup_name)

api/experimentation/metrics.py

@@ -0,0 +1,10 @@
+import prometheus_client
+
+flagsmith_experimentation_warehouse_connection_verifications_total = (
+    prometheus_client.Counter(
+        "flagsmith_experimentation_warehouse_connection_verifications_total",
+        "Outcomes of connection verification attempts against customers' own "
+        "data warehouses. `result` label is either `success` or `failure`.",
+        ["result"],
+    )
+)

api/experimentation/migrations/0010_warehouse_connection_credentials.py

@@ -0,0 +1,17 @@
+from django.db import migrations
+
+import core.fields
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("experimentation", "0009_add_rollout_segment"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="warehouseconnection",
+            name="credentials",
+            field=core.fields.EncryptedJSONField(blank=True, null=True),
+        ),
+    ]

api/experimentation/migrations/0011_warehouse_connection_status_detail.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2.16 on 2026-07-16 09:01
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("experimentation", "0010_warehouse_connection_credentials"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="warehouseconnection",
+            name="status_detail",
+            field=models.CharField(blank=True, max_length=255, null=True),
+        ),
+    ]

api/experimentation/models.py

@@ -12,6 +12,7 @@
     hook,
 )
 
+from core.fields import EncryptedJSONField
 from core.models import SoftDeleteExportableModel
 from environments.models import Environment
 from experimentation.dataclasses import (
@@ -54,10 +55,12 @@ class WarehouseConnection(LifecycleModelMixin, SoftDeleteExportableModel):  # ty
         choices=WarehouseConnectionStatus.choices,
         default=WarehouseConnectionStatus.CREATED,
     )
+    status_detail = models.CharField(max_length=255, null=True, blank=True)
     name = models.CharField(max_length=255)
     config: models.JSONField[dict[str, object] | None, dict[str, object] | None] = (
         models.JSONField(null=True, blank=True)
     )
+    credentials = EncryptedJSONField(null=True, blank=True)
     created_at = models.DateTimeField(auto_now_add=True)
 
     # Populated at serialization time for flagsmith connections from ClickHouse;

api/experimentation/serializers.py

@@ -23,10 +23,10 @@
     apply_experiment_rollout,
     get_experiment_rollout,
 )
-from experimentation.types import (
-    SNOWFLAKE_DEFAULTS,
-    MetricExperimentResult,
-    SnowflakeConfig,
+from experimentation.types import MetricExperimentResult
+from experimentation.warehouse_validation import (
+    CONFIG_VALIDATORS,
+    validate_credentials,
 )
 from features.feature_states.serializers import (
     FeatureValueSerializer,
@@ -41,6 +41,9 @@
 class WarehouseConnectionSerializer(serializers.ModelSerializer):  # type: ignore[type-arg]
     name = serializers.CharField(max_length=255, required=False)
     config = serializers.JSONField(default=None, required=False, allow_null=True)
+    credentials = serializers.JSONField(
+        default=None, required=False, allow_null=True, write_only=True
+    )
     total_events_received = serializers.SerializerMethodField()
     unique_events_count = serializers.SerializerMethodField()
 
@@ -50,27 +53,31 @@ class Meta:
             "id",
             "warehouse_type",
             "status",
+            "status_detail",
             "name",
             "config",
+            "credentials",
             "created_at",
             "total_events_received",
             "unique_events_count",
         )
-        read_only_fields = ("id", "status", "created_at")
+        read_only_fields = ("id", "status", "status_detail", "created_at")
 
     def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
         warehouse_type: str = attrs.get(
             "warehouse_type",
             getattr(self.instance, "warehouse_type", ""),
         )
 
+        validate_credentials(attrs, warehouse_type, self.instance)  # type: ignore[arg-type]
+
         if "config" not in attrs and self.instance is not None:
             return attrs
 
         config: dict[str, Any] | None = attrs.get("config")
 
-        if warehouse_type == WarehouseType.SNOWFLAKE:
-            attrs["config"] = self._validate_snowflake_config(config or {})
+        if config_validator := CONFIG_VALIDATORS.get(warehouse_type):
+            attrs["config"] = config_validator(config or {})
         elif warehouse_type == WarehouseType.FLAGSMITH:
             if config:
                 raise serializers.ValidationError(
@@ -105,19 +112,6 @@ def _generate_name(warehouse_type: str, environment: Environment) -> str:
         label = WarehouseType(warehouse_type).label
         return f"{label} Warehouse - {environment.name}"
 
-    @staticmethod
-    def _validate_snowflake_config(config: dict[str, Any]) -> SnowflakeConfig:
-        account_identifier = config.get("account_identifier", "")
-        if not account_identifier:
-            raise serializers.ValidationError(
-                {"config": {"account_identifier": "This field is required."}}
-            )
-        merged: SnowflakeConfig = {
-            **SNOWFLAKE_DEFAULTS,
-            **config,  # type: ignore[typeddict-item]
-        }
-        return merged
-
 
 class MetricSerializer(serializers.ModelSerializer):  # type: ignore[type-arg]
     experiments = serializers.SerializerMethodField()

api/experimentation/services.py

@@ -6,6 +6,7 @@
 
 import structlog
 from clickhouse_driver import Client
+from clickhouse_driver import errors as clickhouse_errors
 from clickhouse_driver.util.helpers import parse_url
 from django.conf import settings
 from django.db import transaction
@@ -40,6 +41,9 @@
     RolloutSpec,
     WarehouseEventStats,
 )
+from experimentation.metrics import (
+    flagsmith_experimentation_warehouse_connection_verifications_total,
+)
 from experimentation.models import (
     VALID_STATUS_TRANSITIONS,
     ExperimentStatus,
@@ -55,6 +59,7 @@
     compare_to_control,
     srm_p_value,
 )
+from experimentation.types import ClickHouseConfig, ClickHouseCredentials
 from features.models import FeatureState
 from features.value_types import BOOLEAN, INTEGER, STRING
 from features.versioning.dataclasses import FlagChangeSet
@@ -727,6 +732,63 @@ def mark_warehouse_pending_connection(
     return connection
 
 
+def _describe_verification_error(error: Exception) -> str:
+    if isinstance(error, clickhouse_errors.ServerException):
+        if error.code == 516:
+            return "Authentication failed."
+        if error.code == 81:
+            return "Database does not exist."
+        return "The ClickHouse server rejected the request."
+    if isinstance(error, (clickhouse_errors.SocketTimeoutError, TimeoutError)):
+        return "The connection timed out."
+    if isinstance(error, clickhouse_errors.NetworkError):
+        return "Could not connect to the host."
+    if isinstance(error, KeyError):
+        return "Stored connection details are incomplete."
+    return "Verification failed."
+
+
+def verify_clickhouse_connection(connection: WarehouseConnection) -> None:
+    """Run SELECT 1 against the customer's ClickHouse and set the status to
+    connected or errored; never raises."""
+    log = logger.bind(environment__id=connection.environment_id)
+    try:
+        log = log.bind(organisation__id=connection.environment.project.organisation_id)
+        config = typing.cast(ClickHouseConfig, connection.config or {})
+        credentials = typing.cast(ClickHouseCredentials, connection.credentials or {})
+        client = Client(
+            config["host"],
+            port=config["port"],
+            user=config["username"],
+            password=credentials["password"],
+            database=config["database"],
+            secure=config["secure"],
+            connect_timeout=CLICKHOUSE_CONNECT_TIMEOUT_SECONDS,
+            send_receive_timeout=CLICKHOUSE_QUERY_TIMEOUT_SECONDS,
+        )
+        try:
+            client.execute("SELECT 1")
+        finally:
+            client.disconnect()
+    except Exception as error:
+        connection.status = WarehouseConnectionStatus.ERRORED
+        connection.status_detail = _describe_verification_error(error)
+        connection.save(update_fields=["status", "status_detail"])
+        flagsmith_experimentation_warehouse_connection_verifications_total.labels(
+            result="failure"
+        ).inc()
+        log.warning("connection.verification_failed", exc_info=True)
+        return
+
+    connection.status = WarehouseConnectionStatus.CONNECTED
+    connection.status_detail = None
+    connection.save(update_fields=["status", "status_detail"])
+    flagsmith_experimentation_warehouse_connection_verifications_total.labels(
+        result="success"
+    ).inc()
+    log.info("connection.verification_succeeded")
+
+
 def refresh_warehouse_connection_status(
     connection: WarehouseConnection,
     stats: WarehouseEventStats,

api/experimentation/types.py

@@ -44,3 +44,24 @@ class SnowflakeConfig(TypedDict):
     "role": "FLAGSMITH_LOADER",
     "user": "FLAGSMITH_SERVICE",
 }
+
+
+class ClickHouseConfig(TypedDict):
+    host: str
+    port: int
+    database: str
+    username: str
+    secure: bool
+
+
+CLICKHOUSE_DEFAULTS: ClickHouseConfig = {
+    "host": "",
+    "port": 9440,
+    "database": "flagsmith",
+    "username": "default",
+    "secure": True,
+}
+
+
+class ClickHouseCredentials(TypedDict):
+    password: str

api/experimentation/views.py

@@ -15,6 +15,7 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.serializers import BaseSerializer
+from rest_framework.throttling import BaseThrottle, ScopedRateThrottle
 from rest_framework.viewsets import GenericViewSet
 
 from app.pagination import CustomPagination
@@ -62,6 +63,7 @@
     mark_warehouse_pending_connection,
     refresh_warehouse_connection_status,
     transition_experiment_status,
+    verify_clickhouse_connection,
 )
 from experimentation.tasks import (
     compute_experiment_exposures,
@@ -89,19 +91,32 @@ class WarehouseConnectionViewSet(
     lookup_field = "id"
     lookup_url_kwarg = "connection_id"
 
+    def get_throttles(self) -> list[BaseThrottle]:
+        if self.action == "test_warehouse_connection":
+            self.throttle_scope = "warehouse_connection_test"
+            return [ScopedRateThrottle()]
+        return super().get_throttles()
+
     def perform_create(self, serializer: BaseSerializer[WarehouseConnection]) -> None:
         connection: WarehouseConnection = serializer.save(
             environment=self._get_environment()
         )
         create_warehouse_audit_log(
             connection, self._get_user(self.request), action="created"
         )
+        if connection.warehouse_type == WarehouseType.CLICKHOUSE:
+            verify_clickhouse_connection(connection)
 
     def perform_update(self, serializer: BaseSerializer[WarehouseConnection]) -> None:
         connection: WarehouseConnection = serializer.save()
         create_warehouse_audit_log(
             connection, self._get_user(self.request), action="updated"
         )
+        if connection.warehouse_type == WarehouseType.CLICKHOUSE and (
+            "config" in serializer.validated_data
+            or "credentials" in serializer.validated_data
+        ):
+            verify_clickhouse_connection(connection)
 
     def perform_destroy(self, instance: WarehouseConnection) -> None:
         create_warehouse_audit_log(
@@ -130,6 +145,9 @@ def retrieve(self, request: Request, *args: object, **kwargs: object) -> Respons
     @action(detail=True, methods=["post"], url_path="test-warehouse-connection")
     def test_warehouse_connection(self, request: Request, **kwargs: object) -> Response:
         connection: WarehouseConnection = self.get_object()
+        if connection.warehouse_type == WarehouseType.CLICKHOUSE:
+            verify_clickhouse_connection(connection)
+            return Response(self.get_serializer(connection).data)
         if connection.warehouse_type != WarehouseType.FLAGSMITH:
             return Response(
                 {"detail": "Test events are only supported for Flagsmith warehouses."},