feat(cohorts): Amplitude cohort sync endpoints and sync keys

api/api/urls/v1.py

@@ -37,6 +37,7 @@
         r"^multivariate/", include("features.multivariate.urls"), name="multivariate"
     ),
     re_path(r"^segments/", include("segments.urls"), name="segments"),
+    re_path(r"^cohort-sync/", include("cohorts.sync_urls"), name="cohort-sync"),
     re_path(r"^users/", include("users.urls")),
     re_path(r"^e2etests/", include("e2etests.urls")),
     re_path(r"^audit/", include("audit.urls")),

api/cohorts/authentication.py

@@ -0,0 +1,58 @@
+import typing
+from contextlib import suppress
+
+from rest_framework import authentication, exceptions
+from rest_framework.request import Request
+
+from cohorts.models import CohortSyncKey
+
+
+class CohortSyncKeyUser:
+    """Stand-in request user for machine calls authenticated by a
+    CohortSyncKey; carries no permissions of its own."""
+
+    # Named `sync_key`, not `key`: the audit signal that stamps history rows
+    # duck-types master API keys via `request.user.key`.
+    def __init__(self, sync_key: CohortSyncKey) -> None:
+        self.sync_key = sync_key
+
+    def __str__(self) -> str:
+        return self.sync_key.name
+
+    @property
+    def is_authenticated(self) -> bool:
+        return True
+
+    @property
+    def pk(self) -> str:
+        return self.sync_key.id
+
+    @property
+    def is_master_api_key_user(self) -> bool:
+        # The discriminator history/audit code uses for machine callers:
+        # without it, historical records try to store this object as the
+        # acting FFAdminUser and fail.
+        return True
+
+
+class CohortSyncKeyAuthentication(authentication.BaseAuthentication):
+    def authenticate(
+        self, request: Request
+    ) -> tuple[CohortSyncKeyUser, CohortSyncKey] | None:
+        header = request.headers.get("Authorization", "")
+        if not header.startswith("Bearer "):
+            return None
+
+        with suppress(CohortSyncKey.DoesNotExist):
+            key = typing.cast(
+                CohortSyncKey,
+                CohortSyncKey.objects.get_from_key(header.removeprefix("Bearer ")),
+            )
+            if not key.has_expired:
+                return CohortSyncKeyUser(key), key
+
+        raise exceptions.AuthenticationFailed("Valid cohort sync key not found.")
+
+    def authenticate_header(self, request: Request) -> str:
+        # Makes missing credentials a 401 rather than DRF's default 403.
+        return "Bearer"

api/cohorts/migrations/0003_cohort_sync_key.py

@@ -0,0 +1,92 @@
+# Generated by Django 5.2.16 on 2026-08-14 08:31
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cohorts", "0002_cohort_deletion_requested_at"),
+        ("environments", "0039_use_no_ssrf_url_field"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="cohort",
+            name="source_type",
+            field=models.CharField(
+                choices=[("csv", "CSV"), ("amplitude", "Amplitude")],
+                default="csv",
+                max_length=50,
+            ),
+        ),
+        migrations.CreateModel(
+            name="CohortSyncKey",
+            fields=[
+                (
+                    "id",
+                    models.CharField(
+                        editable=False,
+                        max_length=150,
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                    ),
+                ),
+                ("prefix", models.CharField(editable=False, max_length=8, unique=True)),
+                ("hashed_key", models.CharField(editable=False, max_length=150)),
+                ("created", models.DateTimeField(auto_now_add=True, db_index=True)),
+                (
+                    "name",
+                    models.CharField(
+                        default=None,
+                        help_text="A free-form name for the API key. Need not be unique. 50 characters max.",
+                        max_length=50,
+                    ),
+                ),
+                (
+                    "revoked",
+                    models.BooleanField(
+                        blank=True,
+                        default=False,
+                        help_text="If the API key is revoked, clients cannot use it anymore. (This cannot be undone.)",
+                    ),
+                ),
+                (
+                    "expiry_date",
+                    models.DateTimeField(
+                        blank=True,
+                        help_text="Once API key expires, clients cannot use it anymore.",
+                        null=True,
+                        verbose_name="Expires",
+                    ),
+                ),
+                (
+                    "created_by",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "environment",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="cohort_sync_keys",
+                        to="environments.environment",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "API key",
+                "verbose_name_plural": "API keys",
+                "ordering": ("-created",),
+                "abstract": False,
+            },
+        ),
+    ]

api/cohorts/models.py

@@ -1,11 +1,13 @@
 from django.db import models
+from rest_framework_api_key.models import AbstractAPIKey, APIKeyManager
 
 from cohorts.constants import COHORT_SYSTEM_TRAIT_KEY_PREFIX
 from core.models import SoftDeleteExportableModel
 
 
 class CohortSourceType(models.TextChoices):
     CSV = "csv", "CSV"
+    AMPLITUDE = "amplitude", "Amplitude"
 
 
 class Cohort(SoftDeleteExportableModel):
@@ -46,6 +48,26 @@ class Meta:
         ]
 
 
+class CohortSyncKeyManager(APIKeyManager):
+    pass
+
+
+class CohortSyncKey(AbstractAPIKey):
+    """Bearer credential an external cohort source uses to call the
+    cohort-sync endpoints; scopes every call to one environment."""
+
+    environment = models.ForeignKey(
+        "environments.Environment",
+        on_delete=models.CASCADE,
+        related_name="cohort_sync_keys",
+    )
+    created_by = models.ForeignKey(
+        "users.FFAdminUser", on_delete=models.SET_NULL, null=True, blank=True
+    )
+
+    objects = CohortSyncKeyManager()  # type: ignore[misc]
+
+
 class CohortMembershipState(models.TextChoices):
     PENDING_ADD = "pending_add", "Pending add"
     APPLIED = "applied", "Applied"

api/cohorts/serializers.py

@@ -2,7 +2,7 @@
 
 from rest_framework import serializers
 
-from cohorts.models import Cohort
+from cohorts.models import Cohort, CohortSyncKey
 from cohorts.services import create_cohort
 
 
@@ -33,3 +33,32 @@ def create(self, validated_data: dict[str, typing.Any]) -> Cohort:
             name=segment_data["name"],
             description=segment_data.get("description"),
         )
+
+
+class CohortSyncKeySerializer(serializers.ModelSerializer[CohortSyncKey]):
+    key = serializers.SerializerMethodField()
+
+    class Meta:
+        model = CohortSyncKey
+        fields = ("prefix", "name", "created", "key")
+        read_only_fields = ("prefix", "created")
+
+    def create(self, validated_data: dict[str, typing.Any]) -> CohortSyncKey:
+        key, self._generated_key = CohortSyncKey.objects.create_key(**validated_data)
+        return typing.cast(CohortSyncKey, key)
+
+    def get_key(self, instance: CohortSyncKey) -> str | None:
+        # The plaintext key exists only in the create response; it is
+        # unrecoverable afterwards.
+        return getattr(self, "_generated_key", None)
+
+
+class AmplitudeListSerializer(serializers.Serializer[None]):
+    name = serializers.CharField(max_length=2000)
+
+
+class CohortSyncMembersSerializer(serializers.Serializer[None]):
+    # Child length mirrors CohortMembership.identifier.
+    user_ids = serializers.ListField(
+        child=serializers.CharField(max_length=2000), min_length=1
+    )

api/cohorts/services.py

@@ -8,7 +8,12 @@
 
 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 cohorts.models import (
+    Cohort,
+    CohortMembership,
+    CohortMembershipState,
+    CohortSourceType,
+)
 from core.dataclasses import AuthorData
 from environments.dynamodb import DynamoIdentityWrapper
 from segments.models import Condition, Segment, SegmentManagedBy, SegmentRule
@@ -84,6 +89,7 @@ def create_cohort(
     environment: "Environment",
     name: str,
     description: str | None = None,
+    source_type: CohortSourceType = CohortSourceType.CSV,
 ) -> Cohort:
     with transaction.atomic():
         segment = Segment.objects.create(
@@ -93,7 +99,9 @@ def create_cohort(
             managed_by=SegmentManagedBy.COHORT,
         )
         rule = SegmentRule.objects.create(segment=segment, type=SegmentRule.ALL_RULE)
-        cohort: Cohort = Cohort.objects.create(environment=environment, segment=segment)
+        cohort: Cohort = Cohort.objects.create(
+            environment=environment, segment=segment, source_type=source_type
+        )
         Condition.objects.create(
             rule=rule,
             operator=IS_SET,
@@ -115,6 +123,50 @@ def edge_sync_enabled(project: "Project") -> bool:
     return bool(project.enable_dynamo_db and DynamoIdentityWrapper().is_enabled)
 
 
+def add_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None:
+    from cohorts.tasks import apply_cohort_membership_deltas
+
+    rows = [
+        CohortMembership(cohort=cohort, identifier=identifier)
+        for identifier in set(identifiers)
+    ]
+    with transaction.atomic():
+        # Re-adding a member is a no-op end to end: an applied row flips back
+        # to pending and the identity write it triggers is idempotent.
+        CohortMembership.objects.bulk_create(
+            rows,
+            update_conflicts=True,
+            unique_fields=["cohort", "identifier"],
+            update_fields=["state", "updated_at"],
+        )
+        apply_cohort_membership_deltas.delay(kwargs={"cohort_id": cohort.id})
+    logger.info(
+        "membership.deltas_received",
+        cohort__id=cohort.id,
+        environment__id=cohort.environment_id,
+        action="add",
+        deltas__count=len(rows),
+    )
+
+
+def remove_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None:
+    from cohorts.tasks import apply_cohort_membership_deltas
+
+    with transaction.atomic():
+        # Removing a non-member is a no-op: only existing rows flip.
+        updated = CohortMembership.objects.filter(
+            cohort=cohort, identifier__in=set(identifiers)
+        ).update(state=CohortMembershipState.PENDING_REMOVE, updated_at=timezone.now())
+        apply_cohort_membership_deltas.delay(kwargs={"cohort_id": cohort.id})
+    logger.info(
+        "membership.deltas_received",
+        cohort__id=cohort.id,
+        environment__id=cohort.environment_id,
+        action="remove",
+        deltas__count=updated,
+    )
+
+
 def delete_cohort(cohort: Cohort) -> None:
     from cohorts.tasks import apply_cohort_membership_deltas
 

api/cohorts/sync_urls.py

@@ -0,0 +1,10 @@
+from rest_framework.routers import DefaultRouter
+
+from cohorts.sync_views import AmplitudeCohortSyncViewSet
+
+app_name = "cohort-sync"
+
+router = DefaultRouter()
+router.register(r"amplitude/lists", AmplitudeCohortSyncViewSet, basename="amplitude")
+
+urlpatterns = router.urls