Skip to content

Notifier

Module with notification terminator.

Notifier

Bases: ABC

Abstract Notification class.

Source code in mkdocs/lakehouse_engine/packages/terminators/notifier.py 
19
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
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

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

        Args:
            template_field: Message with templates to be replaced.

        Returns:
            Rendered field
        """
        args = {}
        field_template = Template(template_field)
        if (
            NotificationRuntimeParameters.DATABRICKS_JOB_NAME.value in template_field
            or NotificationRuntimeParameters.DATABRICKS_WORKSPACE_ID.value
            in template_field
            or NotificationRuntimeParameters.JOB_EXCEPTION.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
            args["exception"] = self.notification.get("exception")

        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 
24
25
26
27
28
29
30
31
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 
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
@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 
33
34
35
36
@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 
38
39
40
41
@abstractmethod
def send_notification(self) -> None:
    """Abstract send notification method."""
    raise NotImplementedError