ref(protos): Share get_trace_item_type_name for indexed-column item types

Move get_trace_item_type_name from deletion_settings to snuba.protos.common so
the indexed-column lookup can reuse it instead of keeping its own item type
name mapping. Its tests move along with it.

Author: dezzeddi

snuba/datasets/deletion_settings.py

@@ -2,9 +2,6 @@
 
 from collections.abc import Sequence
 from dataclasses import dataclass, field
-from typing import cast
-
-from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
 
 MAX_ROWS_TO_DELETE_DEFAULT = 100000
 
@@ -18,37 +15,3 @@ class DeletionSettings:
     max_rows_to_delete: int = MAX_ROWS_TO_DELETE_DEFAULT
     allowed_attributes_by_item_type: dict[str, list[str]] = field(default_factory=dict)
     partition_column: str | None = None
-
-
-def get_trace_item_type_name(item_type: int) -> str:
-    """
-    Get the string name for a TraceItemType enum value.
-
-    Uses the protobuf enum's Name() method and strips the "TRACE_ITEM_TYPE_" prefix,
-    then converts to lowercase to match storage configuration naming conventions.
-
-    Args:
-        item_type: The integer value of the TraceItemType enum
-
-    Returns:
-        The string name used in storage configurations (e.g., "occurrence", "span")
-
-    Raises:
-        ValueError: If the item_type is not a valid TraceItemType enum value
-    """
-    try:
-        # Get the full protobuf enum name (e.g., "TRACE_ITEM_TYPE_SPAN")
-        # Cast to TraceItemType.ValueType to satisfy type checker
-        full_name = TraceItemType.Name(cast("TraceItemType.ValueType", item_type))
-
-        # Strip the "TRACE_ITEM_TYPE_" prefix and convert to lowercase
-        prefix = "TRACE_ITEM_TYPE_"
-        if not full_name.startswith(prefix):
-            raise ValueError(f"Unexpected TraceItemType name format: {full_name}")
-
-        return full_name[len(prefix) :].lower()
-    except ValueError as e:
-        # This happens when item_type is not a valid enum value
-        raise ValueError(
-            f"Unknown TraceItemType value: {item_type}. Must be a valid TraceItemType enum value."
-        ) from e

snuba/protos/common.py

@@ -1,8 +1,9 @@
 from collections import defaultdict
 from collections.abc import Mapping, Sequence
-from typing import Final, NamedTuple
+from typing import Final, NamedTuple, cast
 
 from sentry_conventions.attributes import ATTRIBUTE_METADATA
+from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
 from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey
 
 from snuba.query.dsl import Functions as f
@@ -393,3 +394,37 @@ def attribute_key_to_expression(attr_key: AttributeKey) -> Expression:
     raise MalformedAttributeException(
         f"Attribute {attr_key.name} has an unknown type: {AttributeKey.Type.Name(attr_key.type)}"
     )
+
+
+def get_trace_item_type_name(item_type: int) -> str:
+    """
+    Get the string name for a TraceItemType enum value.
+
+    Uses the protobuf enum's Name() method and strips the "TRACE_ITEM_TYPE_" prefix,
+    then converts to lowercase to match storage configuration naming conventions.
+
+    Args:
+        item_type: The integer value of the TraceItemType enum
+
+    Returns:
+        The string name used in storage configurations (e.g., "occurrence", "span")
+
+    Raises:
+        ValueError: If the item_type is not a valid TraceItemType enum value
+    """
+    try:
+        # Get the full protobuf enum name (e.g., "TRACE_ITEM_TYPE_SPAN")
+        # Cast to TraceItemType.ValueType to satisfy type checker
+        full_name = TraceItemType.Name(cast("TraceItemType.ValueType", item_type))
+
+        # Strip the "TRACE_ITEM_TYPE_" prefix and convert to lowercase
+        prefix = "TRACE_ITEM_TYPE_"
+        if not full_name.startswith(prefix):
+            raise ValueError(f"Unexpected TraceItemType name format: {full_name}")
+
+        return full_name[len(prefix) :].lower()
+    except ValueError as e:
+        # This happens when item_type is not a valid enum value
+        raise ValueError(
+            f"Unknown TraceItemType value: {item_type}. Must be a valid TraceItemType enum value."
+        ) from e

snuba/web/bulk_delete_query.py

@@ -14,12 +14,13 @@
 from snuba.clickhouse.query import Query
 from snuba.clusters.cluster import UndefinedClickhouseCluster, get_cluster
 from snuba.clusters.storage_sets import StorageSetKey
-from snuba.datasets.deletion_settings import DeletionSettings, get_trace_item_type_name
+from snuba.datasets.deletion_settings import DeletionSettings
 from snuba.datasets.schemas.tables import TableSchema
 from snuba.datasets.storage import WritableTableStorage
 from snuba.datasets.storages.factory import get_config_built_storages
 from snuba.datasets.storages.storage_key import StorageKey
 from snuba.lw_deletions.types import AttributeConditions, ConditionsBag, ConditionsType
+from snuba.protos.common import get_trace_item_type_name
 from snuba.query.conditions import combine_or_conditions
 from snuba.query.data_source.simple import Table
 from snuba.query.dsl import literal

snuba/web/rpc/common/common.py

@@ -28,6 +28,7 @@
     array_element_column,
     coalesced_attribute_names,
     first_present_value,
+    get_trace_item_type_name,
     key_existence_conditions,
     sentry_column,
     type_array_to_membership_array_expression_from_typed_columns,
@@ -1025,23 +1026,6 @@ class IndexedColumn(TypedDict):
 INDEXED_COLUMNS_OPTION = "indexed_columns"
 
 
-_ITEM_TYPE_NAMES: dict[TraceItemType.ValueType, str] = {
-    TraceItemType.TRACE_ITEM_TYPE_SPAN: "span",
-    TraceItemType.TRACE_ITEM_TYPE_ERROR: "error",
-    TraceItemType.TRACE_ITEM_TYPE_LOG: "log",
-    TraceItemType.TRACE_ITEM_TYPE_UPTIME_CHECK: "uptime_check",
-    TraceItemType.TRACE_ITEM_TYPE_UPTIME_RESULT: "uptime_result",
-    TraceItemType.TRACE_ITEM_TYPE_REPLAY: "replay",
-    TraceItemType.TRACE_ITEM_TYPE_OCCURRENCE: "occurrence",
-    TraceItemType.TRACE_ITEM_TYPE_METRIC: "metric",
-    TraceItemType.TRACE_ITEM_TYPE_PROFILE_FUNCTION: "profile_function",
-    TraceItemType.TRACE_ITEM_TYPE_ATTACHMENT: "attachment",
-    TraceItemType.TRACE_ITEM_TYPE_PREPROD: "preprod",
-    TraceItemType.TRACE_ITEM_TYPE_USER_SESSION: "user_session",
-    TraceItemType.TRACE_ITEM_TYPE_PROCESSING_ERROR: "processing_error",
-}
-
-
 def indexed_column_for(
     start_timestamp: ProtobufTimestamp | None,
     item_type: TraceItemType.ValueType,
@@ -1057,7 +1041,7 @@ def indexed_column_for(
     )
     if (indexed_column := indexed_columns.get(unindexed_column)) is None:
         return None
-    if indexed_column.get("item_type") != _ITEM_TYPE_NAMES.get(item_type):
+    if indexed_column.get("item_type") != get_trace_item_type_name(item_type):
         return None
     if start_timestamp.seconds < indexed_column.get("indexed_start_timestamp", 2**63 - 1):
         return None

tests/datasets/test_deletion_settings.py

@@ -1,12 +1,14 @@
 import pytest
 from sentry_conventions.attributes import ATTRIBUTE_METADATA
+from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
 from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey
 
 from snuba.protos.common import (
     ATTRIBUTES_TO_COALESCE,
     MalformedAttributeException,
     _resolve_canonical,
     attribute_key_to_expression,
+    get_trace_item_type_name,
 )
 from snuba.query.dsl import Functions as f
 from snuba.query.dsl import arrayElement, column, literal, map_key_exists
@@ -157,3 +159,33 @@ def test_invalid_type_for_normalized_column_raises_exception(self) -> None:
                 AttributeKey(type=AttributeKey.TYPE_BOOLEAN, name="sentry.trace_id")
             )
         assert "must be one of" in str(exc_info.value)
+
+
+def test_get_trace_item_type_name_valid() -> None:
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_UNSPECIFIED) == "unspecified"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_SPAN) == "span"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_ERROR) == "error"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_LOG) == "log"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_UPTIME_CHECK) == "uptime_check"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_UPTIME_RESULT) == "uptime_result"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_REPLAY) == "replay"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_OCCURRENCE) == "occurrence"
+    assert get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_METRIC) == "metric"
+    assert (
+        get_trace_item_type_name(TraceItemType.TRACE_ITEM_TYPE_PROFILE_FUNCTION)
+        == "profile_function"
+    )
+
+
+def test_get_trace_item_type_name_by_integer() -> None:
+    assert get_trace_item_type_name(0) == "unspecified"
+    assert get_trace_item_type_name(1) == "span"
+    assert get_trace_item_type_name(7) == "occurrence"
+
+
+def test_get_trace_item_type_name_invalid() -> None:
+    with pytest.raises(ValueError, match="Unknown TraceItemType value: 999"):
+        get_trace_item_type_name(999)
+
+    with pytest.raises(ValueError, match="Unknown TraceItemType value: -1"):
+        get_trace_item_type_name(-1)