fix(typing): Type pagerduty module (#98075)

Author: Christinarlong

pyproject.toml

@@ -307,7 +307,6 @@ module = [
     "sentry.api.endpoints.organization_releases",
     "sentry.api.paginator",
     "sentry.db.postgres.base",
-    "sentry.integrations.pagerduty.actions.form",
     "sentry.integrations.slack.message_builder.notifications.issues",
     "sentry.integrations.slack.webhooks.event",
     "sentry.issues.search",
@@ -418,6 +417,7 @@ module = [
     "sentry.integrations.jira_server.actions.*",
     "sentry.integrations.jira_server.utils.*",
     "sentry.integrations.models.integration_feature",
+    "sentry.integrations.pagerduty.*",
     "sentry.integrations.project_management.*",
     "sentry.integrations.repository.*",
     "sentry.integrations.services.*",

src/sentry/integrations/base.py

@@ -414,7 +414,7 @@ def update_organization_config(self, data: MutableMapping[str, Any]) -> None:
         if org_integration is not None:
             self.org_integration = org_integration
 
-    def get_config_data(self) -> Mapping[str, str]:
+    def get_config_data(self) -> Mapping[str, Any]:
         if not self.org_integration:
             return {}
         return self.org_integration.config

src/sentry/integrations/pagerduty/actions/form.py

@@ -10,6 +10,7 @@
 from sentry.integrations.pagerduty.metrics import record_event
 from sentry.integrations.services.integration import integration_service
 from sentry.integrations.types import ExternalProviders
+from sentry.utils.forms import set_field_choices
 
 
 def _validate_int_field(field: str, cleaned_data: Mapping[str, Any]) -> int | None:
@@ -29,28 +30,26 @@ class PagerDutyNotifyServiceForm(forms.Form):
     account = forms.ChoiceField(choices=(), widget=forms.Select())
     service = forms.ChoiceField(required=False, choices=(), widget=forms.Select())
 
-    def __init__(self, *args, **kwargs):
-        integrations = [(i.id, i.name) for i in kwargs.pop("integrations")]
-        services = kwargs.pop("services")
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self._integrations = [(i.id, i.name) for i in kwargs.pop("integrations")]
+        self._services = kwargs.pop("services")
 
         super().__init__(*args, **kwargs)
-        if integrations:
-            self.fields["account"].initial = integrations[0][0]
+        if self._integrations:
+            self.fields["account"].initial = self._integrations[0][0]
 
-        self.fields["account"].choices = integrations
-        self.fields["account"].widget.choices = self.fields["account"].choices
+        set_field_choices(self.fields["account"], self._integrations)
 
-        if services:
-            self.fields["service"].initial = services[0][0]
+        if self._services:
+            self.fields["service"].initial = self._services[0][0]
 
-        self.fields["service"].choices = services
-        self.fields["service"].widget.choices = self.fields["service"].choices
+        set_field_choices(self.fields["service"], self._services)
 
     def _validate_service(self, service_id: int, integration_id: int) -> None:
         with record_event(OnCallInteractionType.VALIDATE_SERVICE).capture() as lifecycle:
             params = {
-                "account": dict(self.fields["account"].choices).get(integration_id),
-                "service": dict(self.fields["service"].choices).get(service_id),
+                "account": dict(self._integrations).get(integration_id),
+                "service": dict(self._services).get(service_id),
             }
 
             org_integrations = integration_service.get_organization_integrations(
@@ -77,11 +76,13 @@ def _validate_service(self, service_id: int, integration_id: int) -> None:
 
     def clean(self) -> dict[str, Any] | None:
         cleaned_data = super().clean()
+        if cleaned_data is None:
+            return cleaned_data
 
         integration_id = _validate_int_field("account", cleaned_data)
         service_id = _validate_int_field("service", cleaned_data)
 
-        if service_id:
+        if service_id and integration_id:
             self._validate_service(service_id, integration_id)
 
         return cleaned_data

src/sentry/integrations/pagerduty/actions/notification.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Sequence
-from typing import cast
+from collections.abc import Generator, Sequence
+from typing import Any, TypedDict, cast
 
 import sentry_sdk
 
@@ -16,20 +16,30 @@
 from sentry.integrations.types import IntegrationProviderSlug
 from sentry.models.rule import Rule
 from sentry.rules.actions import IntegrationEventAction
+from sentry.rules.base import CallbackFuture
+from sentry.services.eventstore.models import GroupEvent
 from sentry.shared_integrations.exceptions import ApiError
+from sentry.types.rules import RuleFuture
 from sentry.utils.strings import truncatechars
 
 logger = logging.getLogger("sentry.integrations.pagerduty")
 
 
+class PagerDutyService(TypedDict):
+    id: int
+    integration_key: str
+    service_name: str
+    integration_id: int
+
+
 class PagerDutyNotifyServiceAction(IntegrationEventAction):
     id = "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction"
     label = "Send a notification to PagerDuty account {account} and service {service} with {severity} severity"
     prompt = "Send a PagerDuty notification"
     provider = IntegrationProviderSlug.PAGERDUTY.value
     integration_key = "account"
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.form_fields = {
             "account": {
@@ -49,7 +59,7 @@ def __init__(self, *args, **kwargs):
             },
         }
 
-    def _get_service(self):
+    def _get_service(self) -> PagerDutyService | None:
         oi = self.get_organization_integration()
         if not oi:
             return None
@@ -58,7 +68,9 @@ def _get_service(self):
                 return pds
         return None
 
-    def after(self, event, notification_uuid: str | None = None):
+    def after(
+        self, event: GroupEvent, notification_uuid: str | None = None
+    ) -> Generator[CallbackFuture]:
         integration = self.get_integration()
         log_context = {
             "organization_id": self.project.organization_id,
@@ -79,7 +91,7 @@ def after(self, event, notification_uuid: str | None = None):
             PagerdutySeverity, self.get_option("severity", default=PAGERDUTY_DEFAULT_SEVERITY)
         )
 
-        def send_notification(event, futures):
+        def send_notification(event: GroupEvent, futures: Sequence[RuleFuture]) -> None:
             installation = integration.get_installation(self.project.organization_id)
             try:
                 client = installation.get_keyring_client(self.get_option("service"))
@@ -147,7 +159,7 @@ def get_services(self) -> Sequence[tuple[int, str]]:
             for v in oi.config.get("pagerduty_services", [])
         ]
 
-    def render_label(self):
+    def render_label(self) -> str:
         s = self._get_service()
         if s:
             service_name = s["service_name"]

src/sentry/integrations/pagerduty/integration.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Mapping, Sequence
-from typing import Any
+from collections.abc import Mapping, MutableMapping, Sequence
+from typing import Any, TypedDict
 
 import orjson
 from django.db import router, transaction
@@ -73,6 +73,23 @@
 )
 
 
+class PagerDutyOrganizationConfig(TypedDict):
+    name: str
+    type: str
+    label: str
+    help: str
+    addButtonText: str
+    columnLabels: dict[str, str]
+    columnKeys: list[str]
+    confirmDeleteMessage: str
+
+
+class PagerDutyServiceConfig(TypedDict):
+    service: str
+    integration_key: str
+    id: int
+
+
 class PagerDutyIntegration(IntegrationInstallation):
     def get_keyring_client(self, keyid: int | str) -> PagerDutyClient:
         org_integration = self.org_integration
@@ -89,11 +106,11 @@ def get_keyring_client(self, keyid: int | str) -> PagerDutyClient:
             integration_id=org_integration.integration_id, integration_key=integration_key
         )
 
-    def get_client(self):
+    def get_client(self) -> None:
         raise NotImplementedError("Use get_keyring_client instead.")
 
-    def get_organization_config(self):
-        fields = [
+    def get_organization_config(self) -> list[PagerDutyOrganizationConfig]:
+        return [
             {
                 "name": "service_table",
                 "type": "table",
@@ -106,9 +123,7 @@ def get_organization_config(self):
             }
         ]
 
-        return fields
-
-    def update_organization_config(self, data):
+    def update_organization_config(self, data: MutableMapping[str, Any]) -> None:
         if "service_table" in data:
             service_rows = data["service_table"]
             # validate fields
@@ -149,15 +164,15 @@ def update_organization_config(self, data):
                     key = row["integration_key"]
                     add_service(oi, integration_key=key, service_name=service_name)
 
-    def get_config_data(self):
+    def get_config_data(self) -> Mapping[str, list[PagerDutyServiceConfig]]:
         service_list = []
         for s in self.services:
             service_list.append(
-                {
-                    "service": s["service_name"],
-                    "integration_key": s["integration_key"],
-                    "id": s["id"],
-                }
+                PagerDutyServiceConfig(
+                    service=s["service_name"],
+                    integration_key=s["integration_key"],
+                    id=s["id"],
+                )
             )
         return {"service_table": service_list}
 
@@ -220,7 +235,7 @@ def build_integration(self, state: Mapping[str, Any]) -> IntegrationData:
 
 
 class PagerDutyInstallationRedirect:
-    def get_app_url(self, account_name=None):
+    def get_app_url(self, account_name: str | None = None) -> str:
         if not account_name:
             account_name = "app"
 

src/sentry/integrations/pagerduty/utils.py

@@ -95,7 +95,7 @@ def build_incident_attachment(
     alert_context: AlertContext,
     metric_issue_context: MetricIssueContext,
     organization: Organization,
-    integration_key,
+    integration_key: str,
     notification_uuid: str | None = None,
 ) -> dict[str, Any]: