ref: fix typing for plugins.bases.issue{,2} (#83445)

<!-- Describe your PR here. -->

Author: asottile-sentry

pyproject.toml

@@ -247,11 +247,8 @@ module = [
     "sentry.pipeline.views.base",
     "sentry.pipeline.views.nested",
     "sentry.plugins.bases.data_forwarding",
-    "sentry.plugins.bases.issue",
-    "sentry.plugins.bases.issue2",
     "sentry.plugins.bases.notify",
     "sentry.plugins.config",
-    "sentry.plugins.endpoints",
     "sentry.receivers.releases",
     "sentry.release_health.metrics_sessions_v2",
     "sentry.replays.endpoints.project_replay_clicks_index",

src/sentry/plugins/bases/issue.py

@@ -24,14 +24,11 @@ class NewIssueForm(forms.Form):
 class IssueTrackingPlugin(Plugin):
     # project_conf_form = BaseIssueOptionsForm
     new_issue_form: type[forms.Form] = NewIssueForm
-    link_issue_form = None
 
     create_issue_template = "sentry/plugins/bases/issue/create_issue.html"
     not_configured_template = "sentry/plugins/bases/issue/not_configured.html"
     needs_auth_template = "sentry/plugins/bases/issue/needs_auth.html"
     auth_provider: str | None = None
-    can_unlink_issues = False
-    can_link_existing_issues = False
 
     def get_plugin_type(self):
         return "issue-tracking"
@@ -58,7 +55,7 @@ def _get_group_title(self, request: Request, group, event):
     def is_configured(self, project) -> bool:
         raise NotImplementedError
 
-    def get_auth_for_user(self, user, **kwargs) -> RpcUserSocialAuth:
+    def get_auth_for_user(self, user, **kwargs) -> RpcUserSocialAuth | None:
         """
         Return a ``RpcUserSocialAuth`` object for the given user based on this plugins ``auth_provider``.
         """
@@ -115,13 +112,6 @@ def get_new_issue_read_only_fields(self, *args, **kwargs):
         """
         return []
 
-    def get_link_existing_issue_form(self, request: Request, group, event, **kwargs):
-        if not self.link_issue_form:
-            return None
-        return self.link_issue_form(
-            request.POST or None, initial=self.get_initial_link_form_data(request, group, event)
-        )
-
     def get_issue_url(self, group, issue_id: str) -> str:
         """
         Given an issue_id (string) return an absolute URL to the issue's details
@@ -161,9 +151,6 @@ def get_initial_form_data(self, request: Request, group, event, **kwargs):
             "title": self._get_group_title(request, group, event),
         }
 
-    def get_initial_link_form_data(self, request: Request, group, event, **kwargs):
-        return {}
-
     def has_auth_configured(self, **kwargs):
         if not self.auth_provider:
             return True
@@ -201,8 +188,6 @@ def view(self, request: Request, group, **kwargs):
             )
 
         if GroupMeta.objects.get_value(group, "%s:tid" % self.get_conf_key(), None):
-            if self.can_unlink_issues and request.GET.get("unlink"):
-                return self.handle_unlink_issue(request, group, **kwargs)
             return None
 
         prefix = self.get_conf_key()
@@ -211,70 +196,40 @@ def view(self, request: Request, group, **kwargs):
         op = request.POST.get("op", "create")
 
         create_form = self.get_new_issue_form(request, group, event)
-        link_form = None
-        if self.can_link_existing_issues:
-            link_form = self.get_link_existing_issue_form(request, group, event)
 
         if op == "create":
-            issue_id = None
             if create_form.is_valid():
                 try:
                     issue_id = self.create_issue(
                         group=group, form_data=create_form.cleaned_data, request=request
                     )
                 except forms.ValidationError as e:
                     create_form.errors["__all__"] = ["Error creating issue: %s" % e]
-
-            if create_form.is_valid():
-                GroupMeta.objects.set_value(group, "%s:tid" % prefix, issue_id)
-
-                issue_information = {
-                    "title": create_form.cleaned_data["title"],
-                    "provider": self.get_title(),
-                    "location": self.get_issue_url(group, issue_id),
-                    "label": self.get_issue_label(group=group, issue_id=issue_id),
-                }
-                Activity.objects.create(
-                    project=group.project,
-                    group=group,
-                    type=ActivityType.CREATE_ISSUE.value,
-                    user_id=request.user.id,
-                    data=issue_information,
-                )
-
-                issue_tracker_used.send_robust(
-                    plugin=self,
-                    project=group.project,
-                    user=request.user,
-                    sender=IssueTrackingPlugin,
-                )
-                return self.redirect(group.get_absolute_url())
-
-        elif op == "link":
-            if link_form.is_valid():
-                try:
-                    self.link_issue(group=group, form_data=link_form.cleaned_data, request=request)
-                except forms.ValidationError as e:
-                    link_form.errors["__all__"] = ["Error creating issue: %s" % e]
-
-            if link_form.is_valid():
-                issue_id = int(link_form.cleaned_data["issue_id"])
-                GroupMeta.objects.set_value(group, "%s:tid" % prefix, issue_id)
-                issue_information = {
-                    "title": self.get_issue_title_by_id(request, group, issue_id),
-                    "provider": self.get_title(),
-                    "location": self.get_issue_url(group, issue_id),
-                    "label": self.get_issue_label(group=group, issue_id=issue_id),
-                }
-                Activity.objects.create(
-                    project=group.project,
-                    group=group,
-                    type=ActivityType.CREATE_ISSUE.value,
-                    user_id=request.user.id,
-                    data=issue_information,
-                )
-
-                return self.redirect(group.get_absolute_url())
+                else:
+                    if create_form.is_valid():
+                        GroupMeta.objects.set_value(group, "%s:tid" % prefix, issue_id)
+
+                        issue_information = {
+                            "title": create_form.cleaned_data["title"],
+                            "provider": self.get_title(),
+                            "location": self.get_issue_url(group, issue_id),
+                            "label": self.get_issue_label(group=group, issue_id=issue_id),
+                        }
+                        Activity.objects.create(
+                            project=group.project,
+                            group=group,
+                            type=ActivityType.CREATE_ISSUE.value,
+                            user_id=request.user.id,
+                            data=issue_information,
+                        )
+
+                        issue_tracker_used.send_robust(
+                            plugin=self,
+                            project=group.project,
+                            user=request.user,
+                            sender=IssueTrackingPlugin,
+                        )
+                        return self.redirect(group.get_absolute_url())
 
         context = {
             "create_form": create_form,
@@ -283,7 +238,6 @@ def view(self, request: Request, group, **kwargs):
             "title": self.get_new_issue_title(),
             "read_only_fields": self.get_new_issue_read_only_fields(group=group),
             "can_link_existing_issues": self.can_link_existing_issues,
-            "link_form": link_form,
             "op": op,
         }
 
@@ -295,10 +249,6 @@ def actions(self, request: Request, group, action_list, **kwargs):
         prefix = self.get_conf_key()
         if not GroupMeta.objects.get_value(group, "%s:tid" % prefix, None):
             action_list.append((self.get_new_issue_title(), self.get_url(group)))
-        elif self.can_unlink_issues:
-            action_list.append(
-                (self.get_unlink_issue_title(), "%s?unlink=1" % self.get_url(group).rstrip("/"))
-            )
         return action_list
 
     def tags(self, request: Request, group, tag_list, **kwargs):

src/sentry/plugins/bases/issue2.py

@@ -1,20 +1,25 @@
 from __future__ import annotations
 
+from collections.abc import Callable
+from typing import TYPE_CHECKING, NotRequired, TypedDict
+
 from django.conf import settings
+from django.http.response import HttpResponseBase
 from django.urls import re_path, reverse
 from rest_framework.request import Request
 from rest_framework.response import Response
 
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
+from sentry.api.bases.group import GroupEndpoint
 from sentry.api.serializers.models.plugin import PluginSerializer
 
 # api compat
 from sentry.exceptions import PluginError  # NOQA
 from sentry.models.activity import Activity
+from sentry.models.group import Group
 from sentry.models.groupmeta import GroupMeta
 from sentry.plugins.base.v1 import Plugin
-from sentry.plugins.endpoints import PluginGroupEndpoint
 from sentry.signals import issue_tracker_used
 from sentry.types.activity import ActivityType
 from sentry.users.services.usersocialauth.model import RpcUserSocialAuth
@@ -23,6 +28,45 @@
 from sentry.utils.http import absolute_uri
 from sentry.utils.safe import safe_execute
 
+if TYPE_CHECKING:
+    from django.utils.functional import _StrPromise
+
+
+@region_silo_endpoint
+class PluginGroupEndpoint(GroupEndpoint):
+    publish_status = {
+        "GET": ApiPublishStatus.PRIVATE,
+        "POST": ApiPublishStatus.PRIVATE,
+    }
+    view: Callable[[Request, Group], HttpResponseBase] = None  # type: ignore[assignment]  # populated by .as_view
+
+    def _handle(self, request: Request, group, *args, **kwargs):
+        GroupMeta.objects.populate_cache([group])
+
+        return self.view(request, group, *args, **kwargs)
+
+    def get(self, request: Request, group, *args, **kwargs) -> Response:
+        return self._handle(request, group, *args, **kwargs)
+
+    def post(self, request: Request, group, *args, **kwargs) -> Response:
+        return self._handle(request, group, *args, **kwargs)
+
+    def respond(self, *args, **kwargs):
+        return Response(*args, **kwargs)
+
+
+class _PluginIssueIssue(TypedDict):
+    issue_id: int
+    url: str
+    label: str
+
+
+class _PluginIssue(TypedDict):
+    slug: str
+    allowed_actions: tuple[str, ...]
+    title: str | _StrPromise
+    issue: NotRequired[_PluginIssueIssue]
+
 
 # TODO(dcramer): remove this in favor of GroupEndpoint
 @region_silo_endpoint
@@ -31,8 +75,8 @@ class IssueGroupActionEndpoint(PluginGroupEndpoint):
         "GET": ApiPublishStatus.PRIVATE,
         "POST": ApiPublishStatus.PRIVATE,
     }
-    view_method_name = None
-    plugin = None
+    view_method_name: str = None  # type: ignore[assignment]  # populated by .as_view
+    plugin: IssuePlugin2 = None  # type: ignore[assignment]  # populated by .as_view
 
     def _handle(self, request: Request, group, *args, **kwargs):
         GroupMeta.objects.populate_cache([group])
@@ -90,7 +134,7 @@ def get_group_urls(self):
             )
         return _urls
 
-    def get_auth_for_user(self, user, **kwargs) -> RpcUserSocialAuth:
+    def get_auth_for_user(self, user, **kwargs) -> RpcUserSocialAuth | None:
         """
         Return a ``RpcUserSocialAuth`` object for the given user based on this plugins ``auth_provider``.
         """
@@ -99,10 +143,9 @@ def get_auth_for_user(self, user, **kwargs) -> RpcUserSocialAuth:
         if not user.is_authenticated:
             return None
 
-        auth = usersocialauth_service.get_one_or_none(
+        return usersocialauth_service.get_one_or_none(
             filter={"user_id": user.id, "provider": self.auth_provider}
         )
-        return auth
 
     def needs_auth(self, request: Request, project, **kwargs):
         """
@@ -358,7 +401,7 @@ def plugin_issues(self, request: Request, group, plugin_issues, **kwargs) -> Non
         if not self.is_configured(project=group.project):
             return
 
-        item = {
+        item: _PluginIssue = {
             "slug": self.slug,
             "allowed_actions": self.allowed_actions,
             "title": self.get_title(),

src/sentry/plugins/endpoints.py

@@ -74,31 +74,6 @@ def get_auth_url(self, user, **kwargs):
 
         return reverse("socialauth_associate", args=[self.auth_provider])
 
-    def needs_auth(self, user, **kwargs):
-        """
-        Return ``True`` if the authenticated user needs to associate an auth
-        service before performing actions with this provider.
-        """
-        if self.auth_provider is None:
-            return False
-
-        organization = kwargs.get("organization")
-        if organization:
-            ois = integration_service.get_organization_integrations(
-                providers=[self.auth_provider], organization_id=organization.id
-            )
-            has_auth = len(ois) > 0
-            if has_auth:
-                return False
-
-        if not user.is_authenticated:
-            return True
-
-        auths = usersocialauth_service.get_many(
-            filter={"user_id": user.id, "provider": self.auth_provider}
-        )
-        return len(auths) == 0
-
     def get_auth(self, user: RpcUser | User, **kwargs) -> RpcUserSocialAuth | None:
         if self.auth_provider is None:
             return None