Function get_group_progress_states

src/sentry/issues/progress.py:45–129

XXX(malwilley): This is a temporary function to derive progress from Activity records. It will be replaced by a derived property on the group action log when available.

(
    group_ids: Sequence[int],
)

def get_group_progress_states(
    group_ids: Sequence[int],
) -> dict[int, str]:
    """
    XXX(malwilley): This is a temporary function to derive progress from Activity records.
    It will be replaced by a derived property on the group action log when available.
    """

    if not group_ids:
        return {}

    assigned_group_ids = set[int](
        GroupAssignee.objects.filter(group_id__in=group_ids).values_list("group_id", flat=True)
    )

    all_progress_activity_types = [
        t for types in ISSUE_PROGRESS_TO_ACTIVITY_TYPES.values() for t in types
    ]

    rows = (
        Activity.objects.filter(
            group_id__in=group_ids,
            type__in=set(all_progress_activity_types) | {ActivityType.SET_REGRESSION.value},
        )
        .values("group_id")
        .annotate(
            latest_regression=Max(
                "datetime",
                filter=Q(type=ActivityType.SET_REGRESSION.value),
            ),
            latest_diagnosed=Max(
                "datetime",
                filter=Q(type__in=ISSUE_PROGRESS_TO_ACTIVITY_TYPES[IssueProgressState.DIAGNOSED]),
            ),
            latest_fix_proposed=Max(
                "datetime",
                filter=Q(
                    type__in=ISSUE_PROGRESS_TO_ACTIVITY_TYPES[IssueProgressState.FIX_PROPOSED]
                ),
            ),
            latest_fix_applied=Max(
                "datetime",
                filter=Q(type__in=ISSUE_PROGRESS_TO_ACTIVITY_TYPES[IssueProgressState.FIX_APPLIED]),
            ),
        )
    )

    annotation_key_by_state = {
        IssueProgressState.FIX_APPLIED: "latest_fix_applied",
        IssueProgressState.FIX_PROPOSED: "latest_fix_proposed",
        IssueProgressState.DIAGNOSED: "latest_diagnosed",
    }

    result: dict[int, str] = {}
    groups_with_activities = set()

    for row in rows:
        group_id = row["group_id"]