chore(pagerduty): support event type in routing key

keephq · Nov 13, 2024 · f64db8a · f64db8a
1 parent d917d4a
commit f64db8a
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 6 deletions.
diff --git a/keep/providers/pagerduty_provider/pagerduty_provider.py b/keep/providers/pagerduty_provider/pagerduty_provider.py
@@ -319,7 +319,11 @@ def validate_scopes(self):
         return scopes
 
     def _build_alert(
-        self, title: str, alert_body: str, dedup: str
+        self,
+        title: str,
+        alert_body: str,
+        dedup: str,
+        event_type: typing.Literal["trigger", "acknowledge", "resolve"] = "trigger",
     ) -> typing.Dict[str, typing.Any]:
         """
         Builds the payload for an event alert.
@@ -328,13 +332,14 @@ def _build_alert(
             title: Title of alert
             alert_body: UTF-8 string of custom message for alert. Shown in incident body
             dedup: Any string, max 255, characters used to deduplicate alerts
+            event_type: The type of event to send to PagerDuty
 
         Returns:
             Dictionary of alert body for JSON serialization
         """
         return {
             "routing_key": self.authentication_config.routing_key,
-            "event_action": "trigger",
+            "event_action": event_type,
             "dedup_key": dedup,
             "payload": {
                 "summary": title,
@@ -346,22 +351,29 @@ def _build_alert(
             },
         }
 
-    def _send_alert(self, title: str, body: str, dedup: str | None = None):
+    def _send_alert(
+        self,
+        title: str,
+        body: str,
+        dedup: str | None = None,
+        event_type: typing.Literal["trigger", "acknowledge", "resolve"] = "trigger",
+    ):
         """
         Sends PagerDuty Alert
 
         Args:
             title: Title of the alert.
             alert_body: UTF-8 string of custom message for alert. Shown in incident body
             dedup: Any string, max 255, characters used to deduplicate alerts
+            event_type: The type of event to send to PagerDuty
         """
         # If no dedup is given, use epoch timestamp
         if dedup is None:
             dedup = str(datetime.datetime.now().timestamp())
 
         url = "https://events.pagerduty.com/v2/enqueue"
 
-        payload = self._build_alert(title, body, dedup)
+        payload = self._build_alert(title, body, dedup, event_type)
         result = requests.post(url, json=payload)
         result.raise_for_status()
 
@@ -496,6 +508,7 @@ def _notify(
         service_id: str = "",
         requester: str = "",
         incident_id: str = "",
+        event_type: typing.Literal["trigger", "acknowledge", "resolve"] = "trigger",
         **kwargs: dict,
     ):
         """
@@ -507,7 +520,9 @@ def _notify(
             kwargs (dict): The providers with context
         """
         if self.authentication_config.routing_key:
-            return self._send_alert(title, alert_body, dedup=dedup)
+            return self._send_alert(
+                title, alert_body, dedup=dedup, event_type=event_type
+            )
         else:
             return self._trigger_incident(
                 service_id, title, alert_body, requester, incident_id

diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "keep"
-version = "0.28.8"
+version = "0.28.9"
 description = "Alerting. for developers, by developers."
 authors = ["Keep Alerting LTD"]
 readme = "README.md"