Skip to content
Merged
15 changes: 15 additions & 0 deletions api/cohorts/migrations/0005_cohort_last_synced_at.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
from django.db import migrations, models


class Migration(migrations.Migration):
dependencies = [
("cohorts", "0004_mixpanel_source"),
]

operations = [
migrations.AddField(
model_name="cohort",
name="last_synced_at",
field=models.DateTimeField(blank=True, null=True),
),
]
1 change: 1 addition & 0 deletions api/cohorts/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ class Cohort(SoftDeleteExportableModel):
external_id = models.CharField(max_length=255, null=True, blank=True)
Comment thread
Zaimwa9 marked this conversation as resolved.
version = models.PositiveIntegerField(default=0)
created_at = models.DateTimeField(auto_now_add=True)
last_synced_at = models.DateTimeField(null=True, blank=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)
Expand Down
56 changes: 54 additions & 2 deletions api/cohorts/serializers.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
import typing

from django.core.files.uploadedfile import UploadedFile
from django.db.models import Count, Q
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers

from cohorts.constants import COHORT_CSV_MAX_FILE_SIZE_BYTES
from cohorts.exceptions import CsvFileTooLargeError
from cohorts.models import Cohort, CohortSyncKey
from cohorts.models import Cohort, CohortMembershipState, CohortSyncKey
from cohorts.services import create_cohort
from environments.models import Environment
from metadata.serializers import MetadataSerializer, MetadataSerializerMixin
Expand All @@ -19,12 +21,19 @@ class Meta:
model = Segment


class CohortMembershipCountsSerializer(serializers.Serializer): # type: ignore[type-arg]
applied = serializers.IntegerField(min_value=0)
pending_add = serializers.IntegerField(min_value=0)
pending_remove = serializers.IntegerField(min_value=0)


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
)
metadata = MetadataSerializer(required=False, many=True, write_only=True)
membership_counts = serializers.SerializerMethodField()

class Meta:
model = Cohort
Expand All @@ -38,11 +47,41 @@ class Meta:
"source_type",
"version",
"created_at",
"last_synced_at",
"membership_counts",
)
read_only_fields = (
"segment",
"source_type",
"version",
"created_at",
"last_synced_at",
)

@extend_schema_field(CohortMembershipCountsSerializer)
def get_membership_counts(self, cohort: Cohort) -> dict[str, int]:
# Clients derive sync status and progress from these. The viewset
# annotates the counts; a freshly created cohort isn't annotated.
if (applied := getattr(cohort, "applied_count", None)) is not None:
return {
"applied": applied,
"pending_add": getattr(cohort, "pending_add_count", 0),
"pending_remove": getattr(cohort, "pending_remove_count", 0),
}
return cohort.memberships.aggregate(
Comment thread
Zaimwa9 marked this conversation as resolved.
applied=Count("id", filter=Q(state=CohortMembershipState.APPLIED)),
pending_add=Count("id", filter=Q(state=CohortMembershipState.PENDING_ADD)),
pending_remove=Count(
"id", filter=Q(state=CohortMembershipState.PENDING_REMOVE)
),
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
read_only_fields = ("segment", "source_type", "version", "created_at")

def validate(self, attrs: dict[str, typing.Any]) -> dict[str, typing.Any]:
attrs = super().validate(attrs)
if self.instance is not None and "metadata" not in attrs:
# A partial update without metadata must not fail the
# required-metadata check.
return attrs
environment = Environment.objects.get(
api_key=self.context["view"].kwargs["environment_api_key"]
)
Expand All @@ -64,6 +103,19 @@ def create(self, validated_data: dict[str, typing.Any]) -> Cohort:
_SegmentMetadataHandler()._update_metadata(cohort.segment, metadata_data)
return cohort

def update(self, instance: Cohort, validated_data: dict[str, typing.Any]) -> Cohort:
# Only the managed segment's fields are updatable.
metadata_data = validated_data.pop("metadata", None)
segment_data = validated_data.pop("segment", {})
if segment_data:
segment = instance.segment
for field, value in segment_data.items():
setattr(segment, field, value)
segment.save(update_fields=list(segment_data))
Comment thread
coderabbitai[bot] marked this conversation as resolved.
if metadata_data is not None:
_SegmentMetadataHandler()._update_metadata(instance.segment, metadata_data)
return instance


class CohortSyncKeySerializer(serializers.ModelSerializer[CohortSyncKey]):
key = serializers.SerializerMethodField()
Expand Down
6 changes: 4 additions & 2 deletions api/cohorts/services.py
Original file line number Diff line number Diff line change
Expand Up @@ -404,8 +404,10 @@ def sync_cohort_memberships_from_csv(
state=CohortMembershipState.PENDING_REMOVE, updated_at=timezone.now()
)

Cohort.objects.filter(id=cohort.id).update(version=F("version") + 1)
cohort.refresh_from_db(fields=["version"])
Cohort.objects.filter(id=cohort.id).update(
version=F("version") + 1, last_synced_at=timezone.now()
)
cohort.refresh_from_db(fields=["version", "last_synced_at"])
apply_cohort_membership_deltas.delay(kwargs={"cohort_id": cohort.id})

flagsmith_cohorts_csv_syncs_total.inc()
Expand Down
26 changes: 24 additions & 2 deletions api/cohorts/views.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from django.db.models import QuerySet
from django.db.models import Count, Q, QuerySet
from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import mixins, status, viewsets
from rest_framework.decorators import action
Expand All @@ -10,7 +10,7 @@

from api.serializers import ErrorSerializer
from cohorts import services
from cohorts.models import Cohort, CohortSyncKey
from cohorts.models import Cohort, CohortMembershipState, CohortSyncKey
from cohorts.permissions import CohortPermission, CohortPlanPermission
from cohorts.serializers import (
CohortCsvSyncResultSerializer,
Expand All @@ -29,6 +29,11 @@
description="Create a cohort and the managed segment that targets it."
),
retrieve=extend_schema(description="Retrieve a cohort."),
partial_update=extend_schema(
description=(
"Update the cohort's managed segment name, description and metadata."
)
),
destroy=extend_schema(
description=(
"Request cohort deletion. Memberships are drained from identity "
Expand All @@ -42,6 +47,7 @@ class CohortViewSet(
mixins.ListModelMixin,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
):
serializer_class = CohortSerializer
Expand All @@ -50,6 +56,8 @@ class CohortViewSet(
model_class = Cohort
lookup_field = "id"
lookup_url_kwarg = "cohort_id"
# PATCH only: a cohort has no meaningful full replacement.
http_method_names = ["get", "post", "patch", "delete", "head", "options"]

def get_queryset(self) -> QuerySet[Cohort]:
# A cohort awaiting drain-then-delete is already gone from the
Expand All @@ -59,6 +67,20 @@ def get_queryset(self) -> QuerySet[Cohort]:
.get_queryset()
.filter(deletion_requested_at__isnull=True)
.select_related("segment")
.annotate(
applied_count=Count(
"memberships",
filter=Q(memberships__state=CohortMembershipState.APPLIED),
),
pending_add_count=Count(
"memberships",
filter=Q(memberships__state=CohortMembershipState.PENDING_ADD),
),
pending_remove_count=Count(
"memberships",
filter=Q(memberships__state=CohortMembershipState.PENDING_REMOVE),
),
)
.order_by("id")
)

Expand Down
Loading
Loading