feat(api): add experimental feature flag update endpoints

api/api/urls/experiments.py

@@ -0,0 +1,25 @@
+"""
+Experimental API endpoints.
+
+These endpoints are subject to change and should not be considered stable.
+Use at your own risk - breaking changes may occur without prior notice.
+"""
+
+from django.urls import path
+
+from features.feature_states.views import update_flag_v1, update_flag_v2
+
+app_name = "experiments"
+
+urlpatterns = [
+    path(
+        "environments/<str:environment_key>/update-flag-v1/",
+        update_flag_v1,
+        name="update-flag-v1",
+    ),
+    path(
+        "environments/<str:environment_key>/update-flag-v2/",
+        update_flag_v2,
+        name="update-flag-v2",
+    ),
+]

api/app/urls.py

@@ -22,6 +22,10 @@
         ),
         re_path(r"^api/v1/", include("api.urls.v1", namespace="api-v1")),
         re_path(r"^api/v2/", include("api.urls.v2", namespace="api-v2")),
+        re_path(
+            r"^api/experiments/",
+            include("api.urls.experiments", namespace="api-experiments"),
+        ),
         re_path(r"^admin/", admin.site.urls),
         re_path(
             r"^sales-dashboard/",

api/features/feature_states/models.py

@@ -38,3 +38,36 @@ def value(self) -> typing.Union[str, int, bool]:
             self.type,  # type: ignore[arg-type]
             self.string_value,
         )
+
+    def set_value(self, value: str, type_: str) -> None:
+        typed_value: str | int | bool
+        match type_:
+            case "string":
+                typed_value = value
+                field = "string_value"
+                type_const = STRING
+            case "integer":
+                try:
+                    typed_value = int(value)
+                except ValueError:
+                    raise ValueError(f"'{value}' is not a valid integer")
+                field = "integer_value"
+                type_const = INTEGER
+            case "boolean":
+                if value.lower() not in ("true", "false"):
+                    raise ValueError(
+                        f"'{value}' is not a valid boolean (use 'true' or 'false')"
+                    )
+                typed_value = value.lower() == "true"
+                field = "boolean_value"
+                type_const = BOOLEAN
+            case _:
+                raise ValueError(
+                    f"'{type_}' is not a valid type (use 'string', 'integer', or 'boolean')"
+                )
+
+        self.string_value = None
+        self.integer_value = None
+        self.boolean_value = None
+        setattr(self, field, typed_value)
+        self.type = type_const

api/features/feature_states/permissions.py

@@ -0,0 +1,19 @@
+from common.environments.permissions import UPDATE_FEATURE_STATE
+from rest_framework.permissions import BasePermission
+from rest_framework.request import Request
+from rest_framework.views import APIView
+
+from environments.models import Environment
+
+
+class EnvironmentUpdateFeatureStatePermission(BasePermission):
+    def has_permission(self, request: Request, view: APIView) -> bool:
+        environment_key = view.kwargs.get("environment_key")
+        try:
+            environment = Environment.objects.get(api_key=environment_key)
+        except Environment.DoesNotExist:
+            return False
+
+        return request.user.has_environment_permission(  # type: ignore[union-attr]
+            UPDATE_FEATURE_STATE, environment
+        )

api/features/feature_states/serializers.py

@@ -0,0 +1,190 @@
+from rest_framework import serializers
+
+from environments.models import Environment
+from features.models import Feature, FeatureState
+from features.versioning.dataclasses import (
+    FlagChangeSet,
+    FlagChangeSetV2,
+    SegmentOverrideChangeSet,
+)
+from features.versioning.versioning_service import update_flag, update_flag_v2
+from segments.models import Segment
+
+
+class BaseFeatureUpdateSerializer(serializers.Serializer):  # type: ignore[type-arg]
+    @property
+    def environment(self) -> Environment:
+        environment: Environment | None = self.context.get("environment")
+        if not environment:
+            raise serializers.ValidationError("Environment context is required")
+        return environment
+
+    def get_feature(self) -> Feature:
+        feature_data = self.validated_data["feature"]
+        try:
+            feature: Feature = Feature.objects.get(
+                project_id=self.environment.project_id, **feature_data
+            )
+            return feature
+        except Feature.DoesNotExist:
+            raise serializers.ValidationError(
+                f"Feature '{feature_data}' not found in project"
+            )
+
+    def validate_segment_id(self, segment_id: int) -> None:
+        if not Segment.objects.filter(
+            id=segment_id, project_id=self.environment.project_id
+        ).exists():
+            raise serializers.ValidationError(
+                f"Segment with id {segment_id} not found in project"
+            )
+
+
+class FeatureIdentifierSerializer(serializers.Serializer):  # type: ignore[type-arg]
+    name = serializers.CharField(required=False, allow_blank=False)
+    id = serializers.IntegerField(required=False)
+
+    def validate(self, data: dict) -> dict:  # type: ignore[type-arg]
+        if not data.get("name") and not data.get("id"):
+            raise serializers.ValidationError(
+                "Either 'name' or 'id' is required for feature identification"
+            )
+        if data.get("name") and data.get("id"):
+            raise serializers.ValidationError("Provide either 'name' or 'id', not both")
+        return data
+
+
+class FeatureUpdateSegmentDataSerializer(serializers.Serializer):  # type: ignore[type-arg]
+    id = serializers.IntegerField(required=True)
+    priority = serializers.IntegerField(required=False, allow_null=True)
+
+
+class FeatureValueSerializer(serializers.Serializer):  # type: ignore[type-arg]
+    type = serializers.ChoiceField(
+        choices=["integer", "string", "boolean"], required=True
+    )
+    string_value = serializers.CharField(required=True, allow_blank=True)
+
+    def validate(self, data: dict) -> dict:  # type: ignore[type-arg]
+        value_type = data["type"]
+        string_val = data["string_value"]
+
+        if value_type == "integer":
+            try:
+                int(string_val)
+            except ValueError:
+                raise serializers.ValidationError(
+                    f"'{string_val}' is not a valid integer"
+                )
+        elif value_type == "boolean":
+            if string_val.lower() not in ("true", "false"):
+                raise serializers.ValidationError(
+                    f"'{string_val}' is not a valid boolean (use 'true' or 'false')"
+                )
+
+        return data
+
+
+class UpdateFlagSerializer(BaseFeatureUpdateSerializer):
+    feature = FeatureIdentifierSerializer(required=True)
+    segment = FeatureUpdateSegmentDataSerializer(required=False)
+    enabled = serializers.BooleanField(required=True)
+    value = FeatureValueSerializer(required=True)
+
+    def validate_segment(self, value: dict) -> dict:  # type: ignore[type-arg]
+        if value and value.get("id"):
+            self.validate_segment_id(value["id"])
+        return value
+
+    @property
+    def flag_change_set(self) -> FlagChangeSet:
+        validated_data = self.validated_data
+        value_data = validated_data["value"]
+        segment_data = validated_data.get("segment")
+
+        change_set = FlagChangeSet(
+            enabled=validated_data["enabled"],
+            feature_state_value=value_data["string_value"],
+            type_=value_data["type"],
+            segment_id=segment_data.get("id") if segment_data else None,
+            segment_priority=segment_data.get("priority") if segment_data else None,
+        )
+
+        change_set.set_audit_fields_from_request(self.context["request"])
+        return change_set
+
+    def save(self, **kwargs: object) -> FeatureState:
+        feature = self.get_feature()
+        return update_flag(self.environment, feature, self.flag_change_set)
+
+
+class EnvironmentDefaultSerializer(serializers.Serializer):  # type: ignore[type-arg]
+    enabled = serializers.BooleanField(required=True)
+    value = FeatureValueSerializer(required=True)
+
+
+class SegmentOverrideSerializer(serializers.Serializer):  # type: ignore[type-arg]
+    segment_id = serializers.IntegerField(required=True)
+    priority = serializers.IntegerField(required=False, allow_null=True)
+    enabled = serializers.BooleanField(required=True)
+    value = FeatureValueSerializer(required=True)
+
+
+class UpdateFlagV2Serializer(BaseFeatureUpdateSerializer):
+    feature = FeatureIdentifierSerializer(required=True)
+    environment_default = EnvironmentDefaultSerializer(required=True)
+    segment_overrides = SegmentOverrideSerializer(many=True, required=False)
+
+    def validate_segment_overrides(
+        self,
+        value: list[dict],  # type: ignore[type-arg]
+    ) -> list[dict]:  # type: ignore[type-arg]
+        if not value:
+            return value
+
+        segment_ids = [override["segment_id"] for override in value]
+        if len(segment_ids) != len(set(segment_ids)):
+            raise serializers.ValidationError(
+                "Duplicate segment_id values are not allowed"
+            )
+
+        for segment_id in segment_ids:
+            self.validate_segment_id(segment_id)
+
+        return value
+
+    @property
+    def change_set_v2(self) -> FlagChangeSetV2:
+        validated_data = self.validated_data
+
+        env_default = validated_data["environment_default"]
+        env_value_data = env_default["value"]
+
+        segment_overrides_data = validated_data.get("segment_overrides", [])
+        segment_overrides = []
+
+        for override_data in segment_overrides_data:
+            value_data = override_data["value"]
+
+            segment_override = SegmentOverrideChangeSet(
+                segment_id=override_data["segment_id"],
+                enabled=override_data["enabled"],
+                feature_state_value=value_data["string_value"],
+                type_=value_data["type"],
+                priority=override_data.get("priority"),
+            )
+            segment_overrides.append(segment_override)
+
+        change_set = FlagChangeSetV2(
+            environment_default_enabled=env_default["enabled"],
+            environment_default_value=env_value_data["string_value"],
+            environment_default_type=env_value_data["type"],
+            segment_overrides=segment_overrides,
+        )
+
+        change_set.set_audit_fields_from_request(self.context["request"])
+        return change_set
+
+    def save(self, **kwargs: object) -> dict:  # type: ignore[type-arg]
+        feature = self.get_feature()
+        return update_flag_v2(self.environment, feature, self.change_set_v2)