diff --git a/src/sentry/rules/actions/integrations/create_ticket/utils.py b/src/sentry/rules/actions/integrations/create_ticket/utils.py index c47679f66fe4f2..be7bd0631455c6 100644 --- a/src/sentry/rules/actions/integrations/create_ticket/utils.py +++ b/src/sentry/rules/actions/integrations/create_ticket/utils.py @@ -2,6 +2,7 @@ import logging from collections.abc import Callable, Sequence +from typing import Any, cast from rest_framework.response import Response @@ -109,18 +110,18 @@ def create_issue(event: GroupEvent, futures: Sequence[RuleFuture]) -> None: organization = event.group.project.organization for future in futures: - rule_id = future.rule.id - data = future.kwargs.get("data") - provider = future.kwargs.get("provider") - integration_id = future.kwargs.get("integration_id") - generate_footer = future.kwargs.get("generate_footer") + rule_id: int = future.rule.id + data = cast(dict[str, Any], future.kwargs.get("data")) + provider = cast(str, future.kwargs.get("provider")) + integration_id = cast(str, future.kwargs.get("integration_id")) + generate_footer = cast(Callable[[str], str], future.kwargs.get("generate_footer")) # If we invoked this handler from the notification action, we need to replace the rule_id with the legacy_rule_id, so we link notifications correctly action_id = None if features.has("organizations:workflow-engine-trigger-actions", organization): # In the Notification Action, we store the rule_id in the action_id field action_id = rule_id - rule_id = data.get("legacy_rule_id") + rule_id = cast(int, data.get("legacy_rule_id")) integration = integration_service.get_integration( integration_id=integration_id, diff --git a/src/sentry/rules/base.py b/src/sentry/rules/base.py index e6cc4450295966..92e2122a9cf578 100644 --- a/src/sentry/rules/base.py +++ b/src/sentry/rules/base.py @@ -2,9 +2,8 @@ import abc import logging -from collections import namedtuple from collections.abc import Callable, MutableMapping, Sequence -from typing import TYPE_CHECKING, Any, ClassVar +from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple from django import forms @@ -48,10 +47,14 @@ - [ACTION:I want to group events when] [RULE:an event matches [FORM]] """ + # Encapsulates a reference to the callback, including arguments. The `key` # attribute may be specifically used to key the callbacks when they are # collated during rule processing. -CallbackFuture = namedtuple("CallbackFuture", ["callback", "kwargs", "key"]) +class CallbackFuture(NamedTuple): + callback: Callable[[GroupEvent, Sequence[RuleFuture]], None] + kwargs: dict[str, Any] + key: str | None class RuleBase(abc.ABC): diff --git a/src/sentry/sentry_apps/tasks/sentry_apps.py b/src/sentry/sentry_apps/tasks/sentry_apps.py index bbcb14feed7902..b61e88b5024e9c 100644 --- a/src/sentry/sentry_apps/tasks/sentry_apps.py +++ b/src/sentry/sentry_apps/tasks/sentry_apps.py @@ -667,7 +667,7 @@ def notify_sentry_app(event: GroupEvent, futures: Sequence[RuleFuture]): # If the future comes from a rule with a UI component form in the schema, append the issue alert payload # TODO(ecosystem): We need to change this payload format after alerts create issues - id = f.rule.id + id: str | int = f.rule.id # if we are using the new workflow engine, we need to use the legacy rule id # Ignore test notifications diff --git a/src/sentry/types/rules.py b/src/sentry/types/rules.py index b8999ff3b2ec7d..58651e04247a6e 100644 --- a/src/sentry/types/rules.py +++ b/src/sentry/types/rules.py @@ -1,7 +1,12 @@ -from collections import namedtuple from dataclasses import dataclass +from typing import Any, NamedTuple -RuleFuture = namedtuple("RuleFuture", ["rule", "kwargs"]) +from sentry.models.rule import Rule + + +class RuleFuture(NamedTuple): + rule: Rule + kwargs: dict[str, Any] @dataclass