feat(cohorts): add environment cohort CRUD API

api/cohorts/migrations/0002_cohort_deletion_requested_at.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.16 on 2026-08-06 07:03
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cohorts", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="cohort",
+            name="deletion_requested_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+    ]

api/cohorts/models.py

@@ -26,6 +26,9 @@ class Cohort(SoftDeleteExportableModel):
     )
     version = models.PositiveIntegerField(default=0)
     created_at = models.DateTimeField(auto_now_add=True)
+    # Deletion drains memberships from the identity store first; the cohort is
+    # only soft-deleted once drained. This marks it as awaiting that final step.
+    deletion_requested_at = models.DateTimeField(null=True, blank=True)
 
     @property
     def system_trait_key(self) -> str:

api/cohorts/permissions.py

@@ -0,0 +1,55 @@
+from common.environments.permissions import (
+    MANAGE_SEGMENT_OVERRIDES,
+    VIEW_ENVIRONMENT,
+)
+from common.projects.permissions import MANAGE_SEGMENTS
+from rest_framework.permissions import BasePermission
+from rest_framework.request import Request
+from rest_framework.views import APIView
+
+from environments.models import Environment
+from organisations.subscriptions.constants import SubscriptionPlanFamily
+from organisations.subscriptions.permissions import require_minimum_plan
+from users.models import FFAdminUser
+
+_READ_ACTIONS = ("list", "retrieve")
+
+_MinimumStartupPlan = require_minimum_plan(SubscriptionPlanFamily.START_UP)
+
+
+class CohortPlanPermission(_MinimumStartupPlan):  # type: ignore[misc,valid-type]
+    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
+        # The base class reads the organisation from an `organisation` request
+        # param our URLs don't carry; the project provides it instead.
+        return bool(super().has_object_permission(request, view, environment.project))
+
+    def has_object_permission(
+        self, request: Request, view: APIView, obj: object
+    ) -> bool:
+        # DRF hands us a Cohort here, which doesn't carry an organisation;
+        # re-run the environment-based check instead.
+        return self.has_permission(request, view)
+
+
+class CohortPermission(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
+        user: FFAdminUser = request.user  # type: ignore[assignment]
+        if not user.has_environment_permission(VIEW_ENVIRONMENT, environment):
+            return False
+        if getattr(view, "action", None) in _READ_ACTIONS:
+            return True
+        return user.has_environment_permission(
+            MANAGE_SEGMENT_OVERRIDES, environment
+        ) and user.has_project_permission(MANAGE_SEGMENTS, environment.project)

api/cohorts/serializers.py

@@ -0,0 +1,35 @@
+import typing
+
+from rest_framework import serializers
+
+from cohorts.models import Cohort
+from cohorts.services import create_cohort
+
+
+class CohortSerializer(serializers.ModelSerializer[Cohort]):
+    name = serializers.CharField(max_length=2000, source="segment.name")
+    description = serializers.CharField(
+        source="segment.description", required=False, allow_null=True
+    )
+
+    class Meta:
+        model = Cohort
+        fields = (
+            "id",
+            "uuid",
+            "name",
+            "description",
+            "segment",
+            "source_type",
+            "version",
+            "created_at",
+        )
+        read_only_fields = ("segment", "source_type", "version", "created_at")
+
+    def create(self, validated_data: dict[str, typing.Any]) -> Cohort:
+        segment_data = validated_data["segment"]
+        return create_cohort(
+            environment=validated_data["environment"],
+            name=segment_data["name"],
+            description=segment_data.get("description"),
+        )

api/cohorts/services.py

@@ -1,11 +1,22 @@
+import typing
+
 import structlog
+from django.db import transaction
 from django.db.models import QuerySet
 from django.utils import timezone
+from flag_engine.segments.constants import IS_SET
 
 from cohorts.constants import COHORT_MEMBERSHIP_APPLY_BATCH_SIZE
 from cohorts.metrics import flagsmith_cohorts_membership_deltas_applied_total
 from cohorts.models import Cohort, CohortMembership, CohortMembershipState
+from core.dataclasses import AuthorData
 from environments.dynamodb import DynamoIdentityWrapper
+from segments.models import Condition, Segment, SegmentManagedBy, SegmentRule
+from segments.services import delete_segment
+
+if typing.TYPE_CHECKING:
+    from environments.models import Environment
+    from projects.models import Project
 
 logger = structlog.get_logger("cohorts")
 
@@ -66,3 +77,68 @@ def apply_pending_memberships(cohort: Cohort) -> bool:
             removes__count=removed_count,
         )
     return pending_memberships(cohort).exists()
+
+
+def create_cohort(
+    *,
+    environment: "Environment",
+    name: str,
+    description: str | None = None,
+) -> Cohort:
+    with transaction.atomic():
+        segment = Segment.objects.create(
+            name=name,
+            project=environment.project,
+            description=description,
+            managed_by=SegmentManagedBy.COHORT,
+        )
+        rule = SegmentRule.objects.create(segment=segment, type=SegmentRule.ALL_RULE)
+        cohort: Cohort = Cohort.objects.create(environment=environment, segment=segment)
+        Condition.objects.create(
+            rule=rule,
+            operator=IS_SET,
+            property=cohort.system_trait_key,
+            created_with_segment=True,
+        )
+    logger.info(
+        "cohort.created",
+        cohort__id=cohort.id,
+        segment__id=segment.id,
+        environment__id=environment.id,
+        project__id=environment.project_id,
+        organisation__id=environment.project.organisation_id,
+    )
+    return cohort
+
+
+def edge_sync_enabled(project: "Project") -> bool:
+    return bool(project.enable_dynamo_db and DynamoIdentityWrapper().is_enabled)
+
+
+def delete_cohort(cohort: Cohort) -> None:
+    from cohorts.tasks import apply_cohort_membership_deltas
+
+    with transaction.atomic():
+        cohort.deletion_requested_at = timezone.now()
+        cohort.save(update_fields=["deletion_requested_at"])
+        logger.info(
+            "cohort.deletion_requested",
+            cohort__id=cohort.id,
+            environment__id=cohort.environment_id,
+        )
+        CohortMembership.objects.filter(cohort=cohort).update(
+            state=CohortMembershipState.PENDING_REMOVE, updated_at=timezone.now()
+        )
+        apply_cohort_membership_deltas.delay(kwargs={"cohort_id": cohort.id})
+
+
+def finalise_cohort_deletion(cohort: Cohort) -> None:
+    segment = cohort.segment
+    with transaction.atomic():
+        cohort.delete()
+        delete_segment(segment, AuthorData())
+    logger.info(
+        "cohort.deleted",
+        cohort__id=cohort.id,
+        environment__id=cohort.environment_id,
+    )

api/cohorts/tasks.py

@@ -11,7 +11,6 @@
     DYNAMODB_THROTTLING_ERROR_CODES,
 )
 from cohorts.models import Cohort
-from environments.dynamodb import DynamoIdentityWrapper
 
 logger = structlog.get_logger("cohorts")
 
@@ -22,15 +21,14 @@ def apply_cohort_membership_deltas(cohort_id: int) -> None:
     if (cohort := Cohort.objects.filter(id=cohort_id).first()) is None:
         log.info("membership.apply.skipped", reason="cohort_missing")
         return
-    if not (
-        cohort.environment.project.enable_dynamo_db
-        and DynamoIdentityWrapper().is_enabled
-    ):
+    if not services.edge_sync_enabled(cohort.environment.project):
         log.info("membership.apply.skipped", reason="not_edge")
         return
     try:
         for _ in range(COHORT_MEMBERSHIP_APPLY_MAX_BATCHES_PER_RUN):
             if not services.apply_pending_memberships(cohort):
+                if cohort.deletion_requested_at is not None:
+                    services.finalise_cohort_deletion(cohort)
                 return
     except ClientError as exc:
         if exc.response["Error"]["Code"] in DYNAMODB_THROTTLING_ERROR_CODES:

api/cohorts/urls.py

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

api/cohorts/views.py

@@ -0,0 +1,64 @@
+from django.db.models import QuerySet
+from drf_spectacular.utils import extend_schema, extend_schema_view
+from rest_framework import mixins, status
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from cohorts import services
+from cohorts.models import Cohort
+from cohorts.permissions import CohortPermission, CohortPlanPermission
+from cohorts.serializers import CohortSerializer
+from environments.views import NestedEnvironmentViewSet
+from projects.exceptions import DynamoNotEnabledError
+
+
+@extend_schema_view(
+    list=extend_schema(description="List the environment's cohorts."),
+    create=extend_schema(
+        description="Create a cohort and the managed segment that targets it."
+    ),
+    retrieve=extend_schema(description="Retrieve a cohort."),
+    destroy=extend_schema(
+        description=(
+            "Request cohort deletion. Memberships are drained from identity "
+            "data first; the cohort and its segment are deleted once drained."
+        ),
+        responses={202: None},
+    ),
+)
+class CohortViewSet(
+    NestedEnvironmentViewSet[Cohort],
+    mixins.ListModelMixin,
+    mixins.CreateModelMixin,
+    mixins.RetrieveModelMixin,
+    mixins.DestroyModelMixin,
+):
+    serializer_class = CohortSerializer
+    pagination_class = None
+    permission_classes = [IsAuthenticated, CohortPlanPermission, CohortPermission]
+    model_class = Cohort
+    lookup_field = "id"
+    lookup_url_kwarg = "cohort_id"
+
+    def initial(self, request: Request, *args: object, **kwargs: object) -> None:
+        super().initial(request, *args, **kwargs)
+        # Cohorts only sync to edge identities for now; core (Postgres
+        # identities) support comes later.
+        if not services.edge_sync_enabled(self._get_environment().project):
+            raise DynamoNotEnabledError()
+
+    def get_queryset(self) -> QuerySet[Cohort]:
+        # A cohort awaiting drain-then-delete is already gone from the
+        # user's point of view.
+        return (
+            super()
+            .get_queryset()
+            .filter(deletion_requested_at__isnull=True)
+            .select_related("segment")
+            .order_by("id")
+        )
+
+    def destroy(self, request: Request, *args: object, **kwargs: object) -> Response:
+        services.delete_cohort(self.get_object())
+        return Response(status=status.HTTP_202_ACCEPTED)

api/core/workflows_services.py

@@ -8,7 +8,10 @@
 from features.versioning.models import EnvironmentFeatureVersion
 from features.versioning.signals import environment_feature_version_published
 from features.versioning.tasks import trigger_update_version_webhooks
-from features.workflows.core.exceptions import ChangeRequestNotApprovedError
+from features.workflows.core.exceptions import (
+    CannotModifyManagedSegmentError,
+    ChangeRequestNotApprovedError,
+)
 
 if TYPE_CHECKING:
     from features.workflows.core.models import ChangeRequest
@@ -26,6 +29,9 @@ def commit(self, committed_by: "FFAdminUser") -> None:
             raise ChangeRequestNotApprovedError(
                 "Change request has not been approved by all required approvers."
             )
+        # Runs before anything publishes: commit is not atomic as a whole, so
+        # raising any later would leave the change request half-applied.
+        self._validate_segments_are_not_cohort_managed()
 
         self._publish_feature_states()
         self._publish_environment_feature_versions(committed_by)
@@ -106,6 +112,16 @@ def _publish_change_sets(self, published_by: "FFAdminUser") -> None:
         for change_set in self.change_request.change_sets.all():
             change_set.publish(user=published_by)
 
+    def _validate_segments_are_not_cohort_managed(self) -> None:
+        for draft_segment in self.change_request.segments.all():
+            if (
+                live_segment := draft_segment.version_of
+            ) and live_segment.cohorts.exists():
+                raise CannotModifyManagedSegmentError(
+                    "Segments managed by a cohort cannot be changed "
+                    "via a change request."
+                )
+
     @transaction.atomic
     def _publish_segments(self) -> None:
         for draft_segment in self.change_request.segments.all():

api/environments/urls.py

@@ -177,6 +177,10 @@
         "<str:environment_api_key>/warehouse-connections/",
         include("experimentation.urls"),
     ),
+    path(
+        "<str:environment_api_key>/cohorts/",
+        include("cohorts.urls"),
+    ),
     path(
         "<str:environment_api_key>/experiments/",
         include("experimentation.experiment_urls"),

api/features/workflows/core/exceptions.py

@@ -16,3 +16,7 @@ class CannotApproveOwnChangeRequest(FeatureWorkflowError):
 
 class ChangeRequestDeletionError(FeatureWorkflowError):
     status_code = status.HTTP_400_BAD_REQUEST  # type: ignore[assignment]
+
+
+class CannotModifyManagedSegmentError(FeatureWorkflowError):
+    status_code = status.HTTP_400_BAD_REQUEST  # type: ignore[assignment]