chore(slack): correctly type issue message builder

pyproject.toml

@@ -310,7 +310,6 @@ module = [
     "sentry.integrations.slack.actions.form",
     "sentry.integrations.slack.client",
     "sentry.integrations.slack.integration",
-    "sentry.integrations.slack.message_builder.issues",
     "sentry.integrations.slack.message_builder.notifications.digest",
     "sentry.integrations.slack.message_builder.notifications.issues",
     "sentry.integrations.slack.notifications",

src/sentry/integrations/message_builder.py

@@ -20,17 +20,17 @@
 
 
 def format_actor_options(
-    actors: Sequence[Team | RpcUser], use_block_kit: bool = False
+    actors: Sequence[Team | RpcUser], is_slack: bool = False
 ) -> Sequence[Mapping[str, str]]:
     sort_func: Callable[[Mapping[str, str]], Any] = lambda actor: actor["text"]
-    if use_block_kit:
+    if is_slack:
         sort_func = lambda actor: actor["text"]["text"]
-    return sorted((format_actor_option(actor, use_block_kit) for actor in actors), key=sort_func)
+    return sorted((format_actor_option(actor, is_slack) for actor in actors), key=sort_func)
 
 
-def format_actor_option(actor: Team | RpcUser, use_block_kit: bool = False) -> Mapping[str, str]:
+def format_actor_option(actor: Team | RpcUser, is_slack: bool = False) -> Mapping[str, str]:
     if isinstance(actor, RpcUser):
-        if use_block_kit:
+        if is_slack:
             return {
                 "text": {
                     "type": "plain_text",
@@ -40,8 +40,8 @@ def format_actor_option(actor: Team | RpcUser, use_block_kit: bool = False) -> M
             }
 
         return {"text": actor.get_display_name(), "value": f"user:{actor.id}"}
-    if isinstance(actor, Team):
-        if use_block_kit:
+    elif isinstance(actor, Team):
+        if is_slack:
             return {
                 "text": {
                     "type": "plain_text",
@@ -51,8 +51,6 @@ def format_actor_option(actor: Team | RpcUser, use_block_kit: bool = False) -> M
             }
         return {"text": f"#{actor.slug}", "value": f"team:{actor.id}"}
 
-    raise NotImplementedError
-
 
 def build_attachment_title(obj: Group | GroupEvent) -> str:
     ev_metadata = obj.get_event_metadata()

src/sentry/integrations/slack/message_builder/issues.py

@@ -3,15 +3,15 @@
 import logging
 from collections.abc import Mapping, Sequence
 from datetime import datetime
-from typing import Any
+from typing import Any, TypedDict
 
 import orjson
 from django.core.exceptions import ObjectDoesNotExist
 from sentry_relay.processing import parse_release
 
 from sentry import tagstore
 from sentry.api.endpoints.group_details import get_group_global_count
-from sentry.constants import LOG_LEVELS_MAP
+from sentry.constants import LOG_LEVELS
 from sentry.eventstore.models import GroupEvent
 from sentry.identity.services.identity import RpcIdentity, identity_service
 from sentry.integrations.message_builder import (
@@ -109,9 +109,9 @@ def build_assigned_text(identity: RpcIdentity, assignee: str) -> str | None:
     except ObjectDoesNotExist:
         return None
 
-    if actor.is_team:
+    if isinstance(assigned_actor, Team):
         assignee_text = f"#{assigned_actor.slug}"
-    elif actor.is_user:
+    elif isinstance(assigned_actor, RpcUser):
         assignee_identity = identity_service.get_identity(
             filter={
                 "provider_id": identity.idp_id,
@@ -147,40 +147,20 @@ def build_action_text(identity: RpcIdentity, action: MessageAction) -> str | Non
     return f"*Issue {status} by <@{identity.external_id}>*"
 
 
-def build_tag_fields(
-    event_for_tags: Any, tags: set[str] | None = None
-) -> Sequence[Mapping[str, str | bool]]:
-    fields = []
-    if tags:
-        event_tags = event_for_tags.tags if event_for_tags else []
-        for key, value in event_tags:
-            std_key = tagstore.backend.get_standardized_key(key)
-            if std_key not in tags:
-                continue
-
-            labeled_value = tagstore.backend.get_tag_value_label(key, value)
-            fields.append(
-                {
-                    "title": std_key.encode("utf-8"),
-                    "value": labeled_value.encode("utf-8"),
-                    "short": True,
-                }
-            )
-    return fields
-
-
-def format_release_tag(value: str, event: GroupEvent | Group):
+def format_release_tag(value: str, event: GroupEvent | None) -> str:
     """Format the release tag using the short version and make it a link"""
+    if not event:
+        return ""
+
     path = f"/releases/{value}/"
     url = event.project.organization.absolute_url(path)
     release_description = parse_release(value, json_loads=orjson.loads).get("description")
     return f"<{url}|{release_description}>"
 
 
 def get_tags(
-    group: Group,
-    event_for_tags: Any,
-    tags: set[str] | None = None,
+    event_for_tags: GroupEvent | None,
+    tags: set[str] | list[tuple[str]] | None = None,
 ) -> Sequence[Mapping[str, str | bool]]:
     """Get tag keys and values for block kit"""
     fields = []
@@ -243,41 +223,30 @@ def get_context(group: Group) -> str:
     return context_text.rstrip()
 
 
-def get_option_groups_block_kit(group: Group) -> Sequence[Mapping[str, Any]]:
-    all_members = group.project.get_members_as_rpc_users()
-    members = list({m.id: m for m in all_members}.values())
-    teams = group.project.teams.all()
-
-    option_groups = []
-    if teams:
-        team_options = format_actor_options(teams, True)
-        option_groups.append(
-            {"label": {"type": "plain_text", "text": "Teams"}, "options": team_options}
-        )
+class OptionGroup(TypedDict):
+    label: Mapping[str, str]
+    options: Sequence[Mapping[str, Any]]
 
-    if members:
-        member_options = format_actor_options(members, True)
-        option_groups.append(
-            {"label": {"type": "plain_text", "text": "People"}, "options": member_options}
-        )
-    return option_groups
 
-
-def get_group_assignees(group: Group) -> Sequence[Mapping[str, Any]]:
-    """Get teams and users that can be issue assignees for block kit"""
+def get_option_groups(group: Group) -> Sequence[OptionGroup]:
     all_members = group.project.get_members_as_rpc_users()
     members = list({m.id: m for m in all_members}.values())
     teams = group.project.teams.all()
 
     option_groups = []
     if teams:
-        for team in teams:
-            option_groups.append({"label": team.slug, "value": f"team:{team.id}"})
+        team_option_group: OptionGroup = {
+            "label": {"type": "plain_text", "text": "Teams"},
+            "options": format_actor_options(teams, True),
+        }
+        option_groups.append(team_option_group)
 
     if members:
-        for member in members:
-            option_groups.append({"label": member.email, "value": f"user:{member.id}"})
-
+        member_option_group: OptionGroup = {
+            "label": {"type": "plain_text", "text": "People"},
+            "options": format_actor_options(members, True),
+        }
+        option_groups.append(member_option_group)
     return option_groups
 
 
@@ -298,20 +267,23 @@ def get_suggested_assignees(
         logger.info("Skipping suspect committers because release does not exist.")
     except Exception:
         logger.exception("Could not get suspect committers. Continuing execution.")
+
     if suggested_assignees:
         suggested_assignees = dedupe_suggested_assignees(suggested_assignees)
         assignee_texts = []
+
         for assignee in suggested_assignees:
             # skip over any suggested assignees that are the current assignee of the issue, if there is any
-            if assignee.is_user and not (
-                isinstance(current_assignee, RpcUser) and assignee.id == current_assignee.id
-            ):
-                assignee_as_user = assignee.resolve()
-                assignee_texts.append(assignee_as_user.get_display_name())
-            elif assignee.is_team and not (
+            if assignee.is_team and not (
                 isinstance(current_assignee, Team) and assignee.id == current_assignee.id
             ):
                 assignee_texts.append(f"#{assignee.slug}")
+            elif assignee.is_user and not (
+                isinstance(current_assignee, RpcUser) and assignee.id == current_assignee.id
+            ):
+                assignee_as_user = assignee.resolve()
+                if isinstance(assignee_as_user, RpcUser):
+                    assignee_texts.append(assignee_as_user.get_display_name())
         return assignee_texts
     return []
 
@@ -417,7 +389,7 @@ def _assign_button() -> MessageAction:
             label="Select Assignee...",
             type="select",
             selected_options=format_actor_options([assignee], True) if assignee else [],
-            option_groups=get_option_groups_block_kit(group),
+            option_groups=get_option_groups(group),
         )
         return assign_button
 
@@ -477,10 +449,10 @@ def escape_text(self) -> bool:
 
     def get_title_block(
         self,
-        rule_id: int,
-        notification_uuid: str,
         event_or_group: GroupEvent | Group,
         has_action: bool,
+        rule_id: int | None = None,
+        notification_uuid: str | None = None,
     ) -> SlackBlock:
         title_link = get_title_link(
             self.group,
@@ -504,11 +476,7 @@ def get_title_block(
                 else ACTIONED_CATEGORY_TO_EMOJI.get(self.group.issue_category)
             )
         elif is_error_issue:
-            level_text = None
-            for k, v in LOG_LEVELS_MAP.items():
-                if self.group.level == v:
-                    level_text = k
-
+            level_text = LOG_LEVELS[self.group.level]
             title_emoji = LEVEL_TO_EMOJI.get(level_text)
         else:
             title_emoji = CATEGORY_TO_EMOJI.get(self.group.issue_category)
@@ -584,7 +552,8 @@ def build(self, notification_uuid: str | None = None) -> SlackBlock:
         # If an event is unspecified, use the tags of the latest event (if one exists).
         event_for_tags = self.event or self.group.get_latest_event()
 
-        obj = self.event if self.event is not None else self.group
+        event_or_group: Group | GroupEvent = self.event if self.event is not None else self.group
+
         action_text = ""
 
         if not self.issue_details or (self.recipient and self.recipient.is_team):
@@ -605,9 +574,9 @@ def build(self, notification_uuid: str | None = None) -> SlackBlock:
             action_text = get_action_text(self.actions, self.identity)
             has_action = True
 
-        blocks = [self.get_title_block(rule_id, notification_uuid, obj, has_action)]
+        blocks = [self.get_title_block(event_or_group, has_action, rule_id, notification_uuid)]
 
-        if culprit_block := self.get_culprit_block(obj):
+        if culprit_block := self.get_culprit_block(event_or_group):
             blocks.append(culprit_block)
 
         # build up text block
@@ -620,7 +589,7 @@ def build(self, notification_uuid: str | None = None) -> SlackBlock:
             blocks.append(self.get_markdown_block(action_text))
 
         # build tags block
-        tags = get_tags(self.group, event_for_tags, self.tags)
+        tags = get_tags(event_for_tags, self.tags)
         if tags:
             blocks.append(self.get_tags_block(tags))
 
@@ -687,7 +656,7 @@ def build(self, notification_uuid: str | None = None) -> SlackBlock:
 
         return self._build_blocks(
             *blocks,
-            fallback_text=self.build_fallback_text(obj, project.slug),
-            block_id=orjson.dumps(block_id).decode(),
+            fallback_text=self.build_fallback_text(event_or_group, project.slug),
+            block_id=block_id,
             skip_fallback=self.skip_fallback,
         )

src/sentry/integrations/slack/webhooks/options_load.py

@@ -2,7 +2,7 @@
 
 import re
 from collections.abc import Mapping, Sequence
-from typing import Any
+from typing import Any, TypedDict
 
 import orjson
 from rest_framework import status
@@ -20,6 +20,11 @@
 from ..utils import logger
 
 
+class OptionGroup(TypedDict):
+    label: Mapping[str, str]
+    options: Sequence[Mapping[str, Any]]
+
+
 @region_silo_endpoint
 class SlackOptionsLoadEndpoint(Endpoint):
     owner = ApiOwner.ECOSYSTEM
@@ -69,15 +74,17 @@ def get_filtered_option_groups(
 
         option_groups = []
         if filtered_teams:
-            team_options = format_actor_options(filtered_teams, True)
-            option_groups.append(
-                {"label": {"type": "plain_text", "text": "Teams"}, "options": team_options}
-            )
+            team_options_group: OptionGroup = {
+                "label": {"type": "plain_text", "text": "Teams"},
+                "options": format_actor_options(filtered_teams, True),
+            }
+            option_groups.append(team_options_group)
         if filtered_members:
-            member_options = format_actor_options(filtered_members, True)
-            option_groups.append(
-                {"label": {"type": "plain_text", "text": "People"}, "options": member_options}
-            )
+            member_options_group: OptionGroup = {
+                "label": {"type": "plain_text", "text": "People"},
+                "options": format_actor_options(filtered_members, True),
+            }
+            option_groups.append(member_options_group)
         return option_groups
 
     # XXX(isabella): atm this endpoint is used only for the assignment dropdown on issue alerts

src/sentry/notifications/utils/participants.py

@@ -8,6 +8,7 @@
 from django.db.models import Q
 
 from sentry import features
+from sentry.eventstore.models import GroupEvent
 from sentry.integrations.types import ExternalProviders
 from sentry.integrations.utils.providers import get_provider_enum_from_string
 from sentry.models.commit import Commit
@@ -262,7 +263,7 @@ def get_owner_reason(
     return None
 
 
-def get_suspect_commit_users(project: Project, event: Event) -> list[RpcUser]:
+def get_suspect_commit_users(project: Project, event: Event | GroupEvent) -> list[RpcUser]:
     """
     Returns a list of users that are suspect committers for the given event.
 
@@ -285,7 +286,7 @@ def get_suspect_commit_users(project: Project, event: Event) -> list[RpcUser]:
     return [committer for committer in suspect_committers if committer.id in in_project_user_ids]
 
 
-def dedupe_suggested_assignees(suggested_assignees: Iterable[Actor]) -> Iterable[Actor]:
+def dedupe_suggested_assignees(suggested_assignees: Iterable[Actor]) -> list[Actor]:
     return list({assignee.id: assignee for assignee in suggested_assignees}.values())
 
 

src/sentry/utils/committers.py

@@ -293,7 +293,7 @@ def get_event_file_committers(
 
 
 def get_serialized_event_file_committers(
-    project: Project, event: Event, frame_limit: int = 25
+    project: Project, event: Event | GroupEvent, frame_limit: int = 25
 ) -> Sequence[AuthorCommitsSerialized]:
 
     group_owners = GroupOwner.objects.filter(

tests/sentry/integrations/slack/notifications/test_issue_alert.py

@@ -40,8 +40,8 @@
 old_get_tags = get_tags
 
 
-def fake_get_tags(group, event_for_tags, tags):
-    return old_get_tags(group, event_for_tags, None)
+def fake_get_tags(event_for_tags, tags):
+    return old_get_tags(event_for_tags, None)
 
 
 class SlackIssueAlertNotificationTest(SlackActivityNotificationTest, PerformanceIssueTestCase):