Skip to content

Notifier

Module with notification terminator.

Notifier

Bases: ABC

Abstract Notification class.

Source code in mkdocs/lakehouse_engine/packages/terminators/notifier.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 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
class Notifier(ABC):
    """Abstract Notification class."""

    _logger = LoggingHandler(__name__).get_logger()

    def __init__(self, notification_spec: TerminatorSpec):
        """Construct Notification instances.

        Args:
            notification_spec: notification specification.
        """
        self.type = notification_spec.args.get("type")
        self.notification = notification_spec.args

    @abstractmethod
    def create_notification(self) -> None:
        """Abstract create notification method."""
        raise NotImplementedError

    @abstractmethod
    def send_notification(self) -> None:
        """Abstract send notification method."""
        raise NotImplementedError

    @staticmethod
    def _check_args_are_correct(given_args: dict, template_args: list) -> None:
        """Checking if all arguments in template were set.

        Args:
            given_args: args set in the terminator spec.
            template_args: args needed to be set in the template.
        """
        extra_args = []

        for key in given_args.keys():
            if key in template_args:
                template_args.remove(key)
            else:
                extra_args.append(key)

        if set(template_args) - set(NOTIFICATION_RUNTIME_PARAMETERS):
            raise ValueError(
                "The following template args have not been set: "
                + ", ".join(template_args)
            )

        if extra_args:
            Notifier._logger.info(
                "Extra parameters sent to template: " + ", ".join(extra_args)
            )

    @staticmethod
    def _render_notification_field(template_field: str, args: dict) -> str:
        """Render the notification given args.

        Args:
            template_field: Message with templates to be replaced.
            args: key/value pairs to be replaced in the message.

        Returns:
            Rendered field
        """
        field_template = Template(template_field)
        if (
            NotificationRuntimeParameters.DATABRICKS_JOB_NAME.value in template_field
            or NotificationRuntimeParameters.DATABRICKS_WORKSPACE_ID.value
            in template_field
        ):
            workspace_id, job_name = DatabricksUtils.get_databricks_job_information(
                ExecEnv.SESSION
            )
            args["databricks_job_name"] = job_name
            args["databricks_workspace_id"] = workspace_id

        return field_template.render(args)

    @staticmethod
    def check_if_notification_is_failure_notification(
        spec: TerminatorSpec,
    ) -> bool:
        """Check if given notification is a failure notification.

        Args:
            spec: spec to validate if it is a failure notification.

        Returns:
            A boolean telling if the notification is a failure notification
        """
        notification = spec.args
        is_notification_failure_notification: bool = False

        if "template" in notification.keys():
            template: dict = NotificationsTemplates.EMAIL_NOTIFICATIONS_TEMPLATES.get(
                notification["template"], {}
            )

            if template:
                is_notification_failure_notification = notification.get(
                    "on_failure", True
                )
            else:
                raise ValueError(f"""Template {notification["template"]} not found.""")
        else:
            is_notification_failure_notification = notification.get("on_failure", True)

        return is_notification_failure_notification

__init__(notification_spec)

Construct Notification instances.

Parameters:

Name Type Description Default
notification_spec TerminatorSpec

notification specification.

 required
Source code in mkdocs/lakehouse_engine/packages/terminators/notifier.py 
25
26
27
28
29
30
31
32
def __init__(self, notification_spec: TerminatorSpec):
    """Construct Notification instances.

    Args:
        notification_spec: notification specification.
    """
    self.type = notification_spec.args.get("type")
    self.notification = notification_spec.args

check_if_notification_is_failure_notification(spec) staticmethod

Check if given notification is a failure notification.

Parameters:

Name Type Description Default
spec TerminatorSpec

spec to validate if it is a failure notification.

 required

Returns:

Type Description
bool

A boolean telling if the notification is a failure notification
Source code in mkdocs/lakehouse_engine/packages/terminators/notifier.py 
 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
@staticmethod
def check_if_notification_is_failure_notification(
    spec: TerminatorSpec,
) -> bool:
    """Check if given notification is a failure notification.

    Args:
        spec: spec to validate if it is a failure notification.

    Returns:
        A boolean telling if the notification is a failure notification
    """
    notification = spec.args
    is_notification_failure_notification: bool = False

    if "template" in notification.keys():
        template: dict = NotificationsTemplates.EMAIL_NOTIFICATIONS_TEMPLATES.get(
            notification["template"], {}
        )

        if template:
            is_notification_failure_notification = notification.get(
                "on_failure", True
            )
        else:
            raise ValueError(f"""Template {notification["template"]} not found.""")
    else:
        is_notification_failure_notification = notification.get("on_failure", True)

    return is_notification_failure_notification

create_notification() abstractmethod

Abstract create notification method.

Source code in mkdocs/lakehouse_engine/packages/terminators/notifier.py 
34
35
36
37
@abstractmethod
def create_notification(self) -> None:
    """Abstract create notification method."""
    raise NotImplementedError

send_notification() abstractmethod

Abstract send notification method.

Source code in mkdocs/lakehouse_engine/packages/terminators/notifier.py 
39
40
41
42
@abstractmethod
def send_notification(self) -> None:
    """Abstract send notification method."""
    raise NotImplementedError