ref(preprod): Extract shared snapshot status derivation for list and detail endpoints

src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot.py

@@ -7,6 +7,7 @@
 import orjson
 from django.conf import settings
 from django.db import IntegrityError, router, transaction
+from django.utils import timezone
 from rest_framework.request import Request
 from rest_framework.response import Response
 
@@ -38,13 +39,18 @@
     SnapshotDetailsApiResponse,
     SnapshotImageResponse,
 )
+from sentry.preprod.api.models.snapshots.snapshot_status import (
+    SnapshotStatusInput,
+    derive_snapshot_status,
+)
 from sentry.preprod.api.schemas import VCS_ERROR_MESSAGES, VCS_SCHEMA_PROPERTIES
 from sentry.preprod.helpers.deletion import delete_artifacts_and_eap_data
 from sentry.preprod.models import PreprodArtifact, PreprodComparisonApproval
 from sentry.preprod.snapshots.comparison_categorizer import (
     CategorizedComparison,
     categorize_comparison_images,
 )
+from sentry.preprod.snapshots.constants import MISSING_BASE_GRACE_PERIOD_SECONDS
 from sentry.preprod.snapshots.manifest import (
     ComparisonManifest,
     ImageMetadata,
@@ -273,16 +279,15 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) ->
 
         comparison_manifest: ComparisonManifest | None = None
         base_manifest: SnapshotManifest | None = None
-        comparison = (
-            PreprodSnapshotComparison.objects.select_related(
-                "base_snapshot_metrics",
-            )
-            .filter(
-                head_snapshot_metrics=snapshot_metrics,
-                state=PreprodSnapshotComparison.State.SUCCESS,
-            )
+        all_comparisons = list(
+            PreprodSnapshotComparison.objects.select_related("base_snapshot_metrics")
+            .filter(head_snapshot_metrics=snapshot_metrics)
             .order_by("-id")
-            .first()
+        )
+        latest_comparison = all_comparisons[0] if all_comparisons else None
+        comparison = next(
+            (c for c in all_comparisons if c.state == PreprodSnapshotComparison.State.SUCCESS),
+            None,
         )
         if comparison:
             comparison_key = (comparison.extras or {}).get("comparison_key")
@@ -324,6 +329,26 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) ->
             )
         )
 
+        has_base_sha = bool(commit_comparison and commit_comparison.base_sha)
+        artifact_age_seconds = (timezone.now() - artifact.date_added).total_seconds()
+        base_artifact_exists: bool | None = None
+        if latest_comparison is None and has_base_sha and commit_comparison is not None:
+            if artifact_age_seconds > MISSING_BASE_GRACE_PERIOD_SECONDS:
+                assert commit_comparison.base_sha is not None
+                base_artifact_exists = (
+                    find_base_snapshot_artifact(
+                        organization_id=commit_comparison.organization_id,
+                        base_sha=commit_comparison.base_sha,
+                        base_repo_name=commit_comparison.base_repo_name
+                        or commit_comparison.head_repo_name,
+                        project_id=artifact.project_id,
+                        app_id=artifact.app_id,
+                        artifact_type=artifact.artifact_type,
+                        build_configuration=artifact.build_configuration,
+                    )
+                    is not None
+                )
+
         image_list = [
             build_snapshot_image_response(key, metadata, manifest.diff_threshold)
             for key, metadata in sorted(manifest.images.items())
@@ -345,18 +370,18 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) ->
             if comparison is not None:
                 base_artifact_id = str(comparison.base_snapshot_metrics.preprod_artifact_id)
             categorized = CategorizedComparison()
-            pending_or_failed_state = (
-                PreprodSnapshotComparison.objects.filter(
-                    head_snapshot_metrics=snapshot_metrics,
-                    state__in=[
+            pending_or_failed_state = next(
+                (
+                    c.state
+                    for c in all_comparisons
+                    if c.state
+                    in (
                         PreprodSnapshotComparison.State.PENDING,
                         PreprodSnapshotComparison.State.PROCESSING,
                         PreprodSnapshotComparison.State.FAILED,
-                    ],
-                )
-                .values_list("state", flat=True)
-                .order_by("-id")
-                .first()
+                    )
+                ),
+                None,
             )
             if pending_or_failed_state is not None:
                 comparison_state = PreprodSnapshotComparison.State(pending_or_failed_state).name
@@ -452,6 +477,17 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) ->
                 approvers=[],
             )
 
+        sorted_approvals = sorted(all_approvals, key=lambda a: a.id, reverse=True)
+        derived_status = derive_snapshot_status(
+            SnapshotStatusInput(
+                latest_comparison=latest_comparison,
+                latest_approval=sorted_approvals[0] if sorted_approvals else None,
+                has_base_sha=has_base_sha,
+                artifact_age_seconds=artifact_age_seconds,
+                base_artifact_exists=base_artifact_exists,
+            )
+        )
+
         response_data = SnapshotDetailsApiResponse(
             head_artifact_id=str(artifact.id),
             base_artifact_id=base_artifact_id,
@@ -479,6 +515,10 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) ->
             comparison_run_info=run_info,
             approval_info=approval_info,
             diff_threshold=manifest.diff_threshold,
+            comparison_state=derived_status.comparison_state,
+            approval_status=derived_status.approval_status,
+            comparison_error_message=derived_status.comparison_error_message,
+            approvers=approver_list if approved else [],
         ).dict()
 
         if compact:

src/sentry/preprod/api/models/project_preprod_build_details_models.py

@@ -6,6 +6,12 @@
 from django.utils import timezone
 from pydantic import BaseModel, Field
 
+from sentry.preprod.api.models.snapshots.snapshot_status import (
+    ApprovalStatusLiteral,
+    ComparisonStateLiteral,
+    SnapshotStatusInput,
+    derive_snapshot_status,
+)
 from sentry.preprod.build_distribution_utils import (
     get_download_count_for_artifact,
     is_installable_artifact,
@@ -94,16 +100,13 @@ class PostedStatusChecks(BaseModel):
 
 class SnapshotComparisonInfo(BaseModel):
     image_count: int
-    comparison_state: (
-        Literal["pending", "processing", "success", "failed", "waiting_for_base", "no_base_build"]
-        | None
-    ) = None
+    comparison_state: ComparisonStateLiteral | None = None
     comparison_error_message: str | None = None
     images_added: int = 0
     images_removed: int = 0
     images_changed: int = 0
     images_unchanged: int = 0
-    approval_status: Literal["approved", "auto_approved", "requires_approval"] | None = None
+    approval_status: ApprovalStatusLiteral | None = None
 
 
 class SizeInfoSizeMetric(BaseModel):
@@ -292,29 +295,23 @@ def to_snapshot_comparison_info(head_artifact: PreprodArtifact) -> SnapshotCompa
     except PreprodSnapshotMetrics.DoesNotExist:
         return None
 
-    comparison_state = None
-    comparison_error_message = None
-    images_added = 0
-    images_removed = 0
-    images_changed = 0
-    images_unchanged = 0
-
     comparisons = sorted(
         snapshot_metrics.snapshot_comparisons_head_metrics.all(),
         key=lambda c: c.id,
         reverse=True,
     )
     comparison = comparisons[0] if comparisons else None
-    if not comparison:
-        cc = head_artifact.commit_comparison
-        if cc and cc.base_sha:
-            grace_period_expired = (
-                timezone.now() - head_artifact.date_added
-            ).total_seconds() > MISSING_BASE_GRACE_PERIOD_SECONDS
-
-            if (
-                grace_period_expired
-                and find_base_snapshot_artifact(
+
+    cc = head_artifact.commit_comparison
+    has_base_sha = bool(cc and cc.base_sha)
+    artifact_age_seconds = (timezone.now() - head_artifact.date_added).total_seconds()
+
+    base_artifact_exists: bool | None = None
+    if comparison is None and has_base_sha and cc is not None:
+        if artifact_age_seconds > MISSING_BASE_GRACE_PERIOD_SECONDS:
+            assert cc.base_sha is not None
+            base_artifact_exists = (
+                find_base_snapshot_artifact(
                     organization_id=cc.organization_id,
                     base_sha=cc.base_sha,
                     base_repo_name=cc.base_repo_name or cc.head_repo_name,
@@ -323,47 +320,45 @@ def to_snapshot_comparison_info(head_artifact: PreprodArtifact) -> SnapshotCompa
                     artifact_type=head_artifact.artifact_type,
                     build_configuration=head_artifact.build_configuration,
                 )
-                is None
-            ):
-                comparison_state = "no_base_build"
-            else:
-                comparison_state = "waiting_for_base"
-    elif comparison:
-        comparison_state = PreprodSnapshotComparison.State(comparison.state).name.lower()
-        if comparison.state == PreprodSnapshotComparison.State.SUCCESS:
-            images_added = comparison.images_added
-            images_removed = comparison.images_removed
-            images_changed = comparison.images_changed
-            images_unchanged = comparison.images_unchanged
-        elif comparison.state == PreprodSnapshotComparison.State.FAILED:
-            comparison_error_message = comparison.error_message
-
-    approval_status = None
-    # REJECTED is no longer used; all non-APPROVED statuses are treated as requires_approval
+                is not None
+            )
+
     approvals = [
         a
         for a in head_artifact.preprodcomparisonapproval_set.all()
         if a.preprod_feature_type == PreprodComparisonApproval.FeatureType.SNAPSHOTS
     ]
     approvals.sort(key=lambda a: a.id, reverse=True)
-    if approvals:
-        if approvals[0].approval_status == PreprodComparisonApproval.ApprovalStatus.APPROVED:
-            if (approvals[0].extras or {}).get("auto_approval") is True:
-                approval_status = "auto_approved"
-            else:
-                approval_status = "approved"
-        else:
-            approval_status = "requires_approval"
+
+    derived = derive_snapshot_status(
+        SnapshotStatusInput(
+            latest_comparison=comparison,
+            latest_approval=approvals[0] if approvals else None,
+            has_base_sha=has_base_sha,
+            artifact_age_seconds=artifact_age_seconds,
+            base_artifact_exists=base_artifact_exists,
+        )
+    )
+
+    images_added = 0
+    images_removed = 0
+    images_changed = 0
+    images_unchanged = 0
+    if comparison is not None and comparison.state == PreprodSnapshotComparison.State.SUCCESS:
+        images_added = comparison.images_added
+        images_removed = comparison.images_removed
+        images_changed = comparison.images_changed
+        images_unchanged = comparison.images_unchanged
 
     return SnapshotComparisonInfo(
         image_count=snapshot_metrics.image_count,
-        comparison_state=comparison_state,
-        comparison_error_message=comparison_error_message,
+        comparison_state=derived.comparison_state,
+        comparison_error_message=derived.comparison_error_message,
         images_added=images_added,
         images_removed=images_removed,
         images_changed=images_changed,
         images_unchanged=images_unchanged,
-        approval_status=approval_status,
+        approval_status=derived.approval_status,
     )
 
 

src/sentry/preprod/api/models/snapshots/project_preprod_snapshot_models.py

@@ -6,6 +6,10 @@
 from pydantic import BaseModel
 
 from sentry.preprod.api.models.project_preprod_build_details_models import BuildDetailsVcsInfo
+from sentry.preprod.api.models.snapshots.snapshot_status import (
+    ApprovalStatusLiteral,
+    ComparisonStateLiteral,
+)
 from sentry.preprod.models import PreprodArtifact
 
 
@@ -132,5 +136,10 @@ class SnapshotDetailsApiResponse(BaseModel):
 
     diff_threshold: float | None = None
 
+    comparison_state: ComparisonStateLiteral | None = None
+    approval_status: ApprovalStatusLiteral | None = None
+    comparison_error_message: str | None = None
+    approvers: list[SnapshotApprover] = []
+
 
 # TODO: POST request in the future when we migrate away from current schemas

src/sentry/preprod/api/models/snapshots/snapshot_status.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+
+from pydantic import BaseModel
+
+from sentry.preprod.models import PreprodComparisonApproval
+from sentry.preprod.snapshots.constants import MISSING_BASE_GRACE_PERIOD_SECONDS
+from sentry.preprod.snapshots.models import PreprodSnapshotComparison
+
+ComparisonStateLiteral = Literal[
+    "pending", "processing", "success", "failed", "waiting_for_base", "no_base_build"
+]
+ApprovalStatusLiteral = Literal["approved", "auto_approved", "requires_approval"]
+
+
+@dataclass(frozen=True)
+class SnapshotStatusInput:
+    latest_comparison: PreprodSnapshotComparison | None
+    latest_approval: PreprodComparisonApproval | None
+    has_base_sha: bool
+    artifact_age_seconds: float
+    base_artifact_exists: bool | None
+
+
+class SnapshotDerivedStatus(BaseModel):
+    comparison_state: ComparisonStateLiteral | None = None
+    approval_status: ApprovalStatusLiteral | None = None
+    comparison_error_message: str | None = None
+
+
+def derive_snapshot_status(status_input: SnapshotStatusInput) -> SnapshotDerivedStatus:
+    comparison_state = None
+    comparison_error_message = None
+
+    if status_input.latest_comparison is not None:
+        comparison_state = PreprodSnapshotComparison.State(
+            status_input.latest_comparison.state
+        ).name.lower()
+        if status_input.latest_comparison.state == PreprodSnapshotComparison.State.FAILED:
+            comparison_error_message = status_input.latest_comparison.error_message
+    elif status_input.has_base_sha:
+        grace_period_expired = status_input.artifact_age_seconds > MISSING_BASE_GRACE_PERIOD_SECONDS
+        if grace_period_expired and status_input.base_artifact_exists is False:
+            comparison_state = "no_base_build"
+        else:
+            comparison_state = "waiting_for_base"
+
+    approval_status = None
+    if status_input.latest_approval is not None:
+        if (
+            status_input.latest_approval.approval_status
+            == PreprodComparisonApproval.ApprovalStatus.APPROVED
+        ):
+            if (status_input.latest_approval.extras or {}).get("auto_approval") is True:
+                approval_status = "auto_approved"
+            else:
+                approval_status = "approved"
+        else:
+            approval_status = "requires_approval"
+
+    return SnapshotDerivedStatus(
+        comparison_state=comparison_state,
+        comparison_error_message=comparison_error_message,
+        approval_status=approval_status,
+    )