create event payload interface

Author: GabrielSalla

docs/monitor.md

@@ -268,24 +268,21 @@ Reactions are defined as a list of **async functions** that are triggered when s
 Below is an example of defining a reaction function that responds to the creation of a new issue:
 
 ```python
-async def reaction_issue_created(event_payload: dict[str, Any]) -> None:
+from monitor_utils import EventPayload
+
+
+async def reaction_issue_created(event_payload: EventPayload) -> None:
     # Do something
 ```
 
+### Event payload
 The event payload provided to each reaction function contains structured information about the event source, details, and any additional context. This allows reaction functions to respond precisely to specific events.
-
-```python
-{
-    "event_source": "Specifies the model that generated the event (e.g., `monitor`, `issue`, `alert`)."
-    "event_source_id": "The unique identifier of the object that triggered the event (e.g., `monitor_id`, `issue_id`)."
-    "event_source_monitor_id": "The monitor ID associated with the object that generated the event."
-    "event_name": "Name of the event, such as `alert_created` or `issue_solved`.",
-    "event_data": {
-        "Object with detailed information about the event source."
-    },
-    "extra_payload": "Additional information that may be sent along with the event, providing further context.",
-}
-```
+- `event_source`: Specifies the model that generated the event (e.g., `monitor`, `issue`, `alert`).
+- `event_source_id`: The unique identifier of the object that triggered the event (e.g., `monitor_id`, `issue_id`).
+- `event_source_monitor_id`: The monitor ID associated with the object that generated the event.
+- `event_name`: Name of the event, such as `alert_created` or `issue_solved`.
+- `event_data`: Dictionary with detailed information about the event source.
+- `extra_payload`: Additional information that may be sent along with the event, providing further context.
 
 Reaction functions can be assigned to specific events when creating an instance of `ReactionOptions`. This configuration ensures that designated functions are triggered whenever specified events occur.
 

internal_monitors/active_notification_alert_solved/active_notification_alert_solved.py

@@ -5,14 +5,15 @@
 """
 
 import os
-from typing import Any, TypedDict, cast
+from typing import TypedDict, cast
 
 from databases import query_application
 from models import Notification, NotificationStatus
 from monitor_utils import (
     AgeRule,
     AlertOptions,
     AlertPriority,
+    EventPayload,
     IssueOptions,
     MonitorOptions,
     PriorityLevels,
@@ -69,9 +70,9 @@ def is_solved(issue_data: IssueDataType) -> bool:
 # Reactions
 
 
-async def close_notification(event_payload: dict[str, Any]) -> None:
+async def close_notification(event_payload: EventPayload) -> None:
     """Fix the notification by closing it"""
-    issue_object = event_payload["event_data"]
+    issue_object = event_payload.event_data
     notification = await Notification.get_by_id(issue_object["data"]["notification_id"])
     if notification:
         await notification.close()

src/components/executor/reaction_handler.py

@@ -11,6 +11,7 @@
 from base_exception import BaseSentinelaException
 from configs import configs
 from models import Monitor
+from options import EventPayload
 
 _logger = logging.getLogger("reaction_handler")
 
@@ -34,9 +35,9 @@
 async def run(message: dict[Any, Any]) -> None:
     """Process a message with type 'event' using the monitor's defined list of reactions for the
     event. The execution timeout is for each function individually"""
-    message_payload = message["payload"]
-    monitor_id = message_payload["event_source_monitor_id"]
-    event_name = message_payload["event_name"]
+    event_payload = EventPayload(**message["payload"])
+    monitor_id = event_payload.event_source_monitor_id
+    event_name = event_payload.event_name
 
     monitor = await Monitor.get_by_id(monitor_id)
     if monitor is None:
@@ -68,19 +69,19 @@ async def run(message: dict[Any, Any]) -> None:
         reaction_execution_time = prometheus_reaction_execution_time.labels(**prometheus_labels)
         try:
             with reaction_execution_time.time():
-                await asyncio.wait_for(reaction(message_payload), configs.executor_reaction_timeout)
+                await asyncio.wait_for(reaction(event_payload), configs.executor_reaction_timeout)
         except asyncio.TimeoutError:
             prometheus_reaction_timeout_count.labels(**prometheus_labels).inc()
             _logger.error(
                 f"Timed out executing reaction '{reaction_name}' with payload "
-                f"'{json.dumps(message_payload)}'"
+                f"'{json.dumps(event_payload.to_dict())}'"
             )
         except BaseSentinelaException as e:
             raise e
         except Exception:
             prometheus_reaction_error_count.labels(**prometheus_labels).inc()
             _logger.error(
                 f"Error executing reaction '{reaction_name}' with payload "
-                f"'{json.dumps(message_payload)}'"
+                f"'{json.dumps(event_payload.to_dict())}'"
             )
             _logger.error(traceback.format_exc().strip())

src/monitor_utils/__init__.py

@@ -8,6 +8,7 @@
     AgeRule,
     AlertOptions,
     CountRule,
+    EventPayload,
     IssueOptions,
     MonitorOptions,
     PriorityLevels,
@@ -22,6 +23,7 @@
     "AlertOptions",
     "AlertPriority",
     "CountRule",
+    "EventPayload",
     "IssueOptions",
     "MonitorOptions",
     "PriorityLevels",

src/notifications/base_notification.py

@@ -1,10 +1,10 @@
-from typing import Any, Callable, Coroutine, Protocol, runtime_checkable
+from typing import Protocol, runtime_checkable
 
-type async_function = Callable[[dict[str, Any]], Coroutine[Any, Any, Any]]
+from options import reaction_function_type
 
 
 @runtime_checkable
 class BaseNotification(Protocol):
     min_priority_to_send: int = 5
 
-    def reactions_list(self) -> list[tuple[str, list[async_function]]]: ...
+    def reactions_list(self) -> list[tuple[str, list[reaction_function_type]]]: ...

src/options/__init__.py

@@ -2,20 +2,24 @@
     AgeRule,
     AlertOptions,
     CountRule,
+    EventPayload,
     IssueOptions,
     MonitorOptions,
     PriorityLevels,
     ReactionOptions,
     ValueRule,
+    reaction_function_type,
 )
 
 __all__ = [
     "AgeRule",
     "AlertOptions",
     "CountRule",
+    "EventPayload",
     "IssueOptions",
     "MonitorOptions",
     "PriorityLevels",
     "ReactionOptions",
     "ValueRule",
+    "reaction_function_type",
 ]

src/options/options.py

@@ -120,7 +120,40 @@ class AlertOptions:
     dismiss_acknowledge_on_new_issues: bool = False
 
 
-reaction_function_type = Callable[[dict[str, Any]], Coroutine[Any, Any, Any]]
+@dataclass
+class EventPayload:
+    """The event payload provided to each reaction function contains structured information about
+    the event source, details, and any additional context.
+    - `event_source`: Specifies the model that generated the event (e.g., `monitor`, `issue`,
+    `alert`).
+    - `event_source_id`: The unique identifier of the object that triggered the event (e.g.,
+    `monitor_id`, `issue_id`).
+    - `event_source_monitor_id`: The monitor ID associated with the object that generated the event.
+    - `event_name`: Name of the event, such as `alert_created` or `issue_solved`.
+    - `event_data`: Object with detailed information about the event source.
+    - `extra_payload`: Additional information that may be sent along with the event, providing
+    further context.
+    """
+
+    event_source: str
+    event_source_id: int
+    event_source_monitor_id: int
+    event_name: str
+    event_data: dict[str, Any]
+    extra_payload: dict[str, Any] | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "event_source": self.event_source,
+            "event_source_id": self.event_source_id,
+            "event_source_monitor_id": self.event_source_monitor_id,
+            "event_name": self.event_name,
+            "event_data": self.event_data,
+            "extra_payload": self.extra_payload,
+        }
+
+
+reaction_function_type = Callable[[EventPayload], Coroutine[Any, Any, Any]]
 
 
 @dataclass
@@ -137,23 +170,6 @@ class ReactionOptions:
     event source, details, and any additional context. This allows reaction functions to respond
     precisely to specific events.
 
-    ```python
-    {
-        "event_source": "Specifies the model that generated the event (e.g., `monitor`, `issue`,
-        `alert`)."
-        "event_source_id": "The unique identifier of the object that triggered the event (e.g.,
-        `monitor_id`, `issue_id`)."
-        "event_source_monitor_id": "The monitor ID associated with the object that generated the
-        event."
-        "event_name": "Name of the event, such as `alert_created` or `issue_solved`.",
-        "event_data": {
-            "Object with detailed information about the event source."
-        },
-        "extra_payload": "Additional information that may be sent along with the event, providing
-        further context.",
-    }
-    ```
-
     Check the documentation for a more detailed explanation of each event.
     """
 

src/plugins/slack/notifications/slack_notification.py

@@ -2,21 +2,20 @@
 import logging
 import os
 from functools import partial
-from typing import Any, Callable, Coroutine
+from typing import Any
 
 from pydantic.dataclasses import dataclass
 from pytz import timezone
 from tabulate import tabulate
 
 from configs import configs
 from models import Alert, AlertPriority, AlertStatus, Issue, IssueStatus, Monitor, Notification
+from options import EventPayload, reaction_function_type
 
 from .. import slack
 
 _logger = logging.getLogger("plugin.slack.notifications")
 
-type async_function = Callable[[dict[str, Any]], Coroutine[Any, Any, Any]]
-
 RESEND_ERRORS = [
     "message_not_found",
     "cant_update_message",
@@ -56,7 +55,7 @@ class SlackNotification:
     min_priority_to_mention: int = AlertPriority.moderate
     issue_show_limit: int = 10
 
-    def reactions_list(self) -> list[tuple[str, list[async_function]]]:
+    def reactions_list(self) -> list[tuple[str, list[reaction_function_type]]]:
         """Get a list of events that the notification will react to"""
         handle_notification_function = partial(slack_notification, notification_options=self)
         return [
@@ -387,11 +386,11 @@ async def notification_mention(
 
 
 async def slack_notification(
-    event_payload: dict[str, Any],
+    event_payload: EventPayload,
     notification_options: SlackNotification,
 ) -> None:
     """Handle the Slack notification for an alert"""
-    alert_data = event_payload["event_data"]
+    alert_data = event_payload.event_data
 
     alert = await Alert.get_by_id(alert_data["id"])
     if alert is None: