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,48 @@
+from common.environments.permissions import 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 getattr(view, "action", None) in _READ_ACTIONS:
+            return user.has_environment_permission(VIEW_ENVIRONMENT, environment)
+        return user.has_project_permission(MANAGE_SEGMENTS, environment.project)

api/cohorts/serializers.py

@@ -0,0 +1,29 @@
+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")
+
+    class Meta:
+        model = Cohort
+        fields = (
+            "id",
+            "uuid",
+            "name",
+            "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:
+        return create_cohort(
+            environment=validated_data["environment"],
+            name=validated_data["segment"]["name"],
+        )

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,62 @@ def apply_pending_memberships(cohort: Cohort) -> bool:
             removes__count=removed_count,
         )
     return pending_memberships(cohort).exists()
+
+
+def create_cohort(*, environment: "Environment", name: str) -> Cohort:
+    with transaction.atomic():
+        segment = Segment.objects.create(
+            name=name,
+            project=environment.project,
+            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/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/segments/migrations/0031_segment_managed_by.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.2.16 on 2026-08-11 07:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("segments", "0030_add_default_to_segment_version"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="historicalsegment",
+            name="managed_by",
+            field=models.CharField(
+                blank=True, choices=[("cohort", "Cohort")], default="", max_length=50
+            ),
+        ),
+        migrations.AddField(
+            model_name="segment",
+            name="managed_by",
+            field=models.CharField(
+                blank=True, choices=[("cohort", "Cohort")], default="", max_length=50
+            ),
+        ),
+    ]

api/segments/models.py

@@ -77,6 +77,10 @@ class SegmentConditionManager(ConfiguredOrderManager["Condition"]):
     setting_name = "SEGMENT_CONDITIONS_EXPLICIT_ORDERING_ENABLED"
 
 
+class SegmentManagedBy(models.TextChoices):
+    COHORT = "cohort", "Cohort"
+
+
 class Segment(
     LifecycleModelMixin,  # type: ignore[misc]
     SoftDeleteExportableModel,
@@ -121,6 +125,12 @@ class Segment(
     created_at = models.DateTimeField(null=True, auto_now_add=True)
     updated_at = models.DateTimeField(null=True, auto_now=True)
     is_system_segment = models.BooleanField(default=False)
+    # A managed segment is created and maintained by another feature (e.g. a
+    # cohort). Unlike system segments it stays visible in the API, but the
+    # dashboard renders it differently and cannot edit it.
+    managed_by = models.CharField(
+        max_length=50, choices=SegmentManagedBy.choices, default="", blank=True
+    )
 
     objects = SegmentManager()  # type: ignore[misc]
 

api/segments/serializers.py

@@ -126,8 +126,9 @@ class Meta:
             "rules",
             "metadata",
             "membership_counts",
+            "managed_by",
         ]
-        read_only_fields = ["membership_counts"]
+        read_only_fields = ["membership_counts", "managed_by"]
 
     def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
         attrs = super().validate(attrs)

api/segments/views.py

@@ -214,6 +214,17 @@ def members(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         next_cursor = members[-1]["identifier"] if has_more else None
         return Response({"results": members, "next_cursor": next_cursor})
 
+    def check_object_permissions(self, request: Request, obj: Segment) -> None:
+        super().check_object_permissions(request, obj)
+        if (
+            self.action in ("update", "partial_update", "destroy", "clone")
+            and obj.cohorts.exists()
+        ):
+            raise PermissionDenied(
+                "This segment is managed by a cohort and cannot be edited "
+                "or cloned directly."
+            )
+
     def destroy(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         segment = self.get_object()
         author = AuthorData.from_request(request)