Skip to content

Attributes

Classes

spark_expectations.notifications.plugins.pagerduty.SparkExpectationsPagerDutyPluginImpl

Bases: SparkExpectationsNotification

This class implements/supports functionality to send PagerDuty notifications

Functions

send_notification(_context: SparkExpectationsContext, _config_args: Dict[str, Union[str, bool]]) -> None

function to create a PagerDuty incident Args: _context: SparkExpectationsContext class object _config_args: dict which contains message(str) and other related configurations for PD Returns: None

Source code in .venv/env/virtual/dev.py3.12/lib/python3.12/site-packages/spark_expectations/notifications/plugins/pagerduty.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@spark_expectations_notification_impl
def send_notification(
    self,
    _context: SparkExpectationsContext,
    _config_args: Dict[str, Union[str, bool]],
) -> None:
    """
    function to create a PagerDuty incident
    Args:
        _context: SparkExpectationsContext class object
        _config_args: dict which contains message(str) and other related configurations for PD
    Returns: None
    """
    try:
        if _context.get_enable_pagerduty is True:
            message = _config_args.get("message")

            if _context.get_pagerduty_integration_key:
                self._get_pd_integration_key(_context)

            # Sending request to PagerDuty Events API v2 > https://developer.pagerduty.com/docs/send-alert-event
            # Severity Levels can be: critical, error, warning, or info
            payload = {
                "routing_key": _context.get_pagerduty_integration_key,
                "event_action": "trigger",
                "payload": {
                    "summary": message,
                    "source": "Spark Expectations",
                    "severity": "error",
                },
            }
            headers = {
                "Content-Type": "application/json",
            }
            response = requests.post(
                _context.get_pagerduty_webhook_url,
                json=payload,
                headers=headers,
                timeout=10,
            )

            # Check the response for success or failure
            if response.status_code == 202:
                _log.info("PagerDuty notification sent successfully!")
            else:
                raise SparkExpectationsPagerDutyException(
                    f"Failed to send PagerDuty notification. Status code: {response.status_code}, Response: {response.text}"
                )
    except Exception as e:
        raise SparkExpectationsPagerDutyException(f"Error sending PagerDuty notification: {str(e)}")