Skip to content

Send email notifications

Send email notifications for all tasks that have failed.

gather_failed_tasks()

Gather all tasks that have failed.

Source code in gso/workflows/tasks/send_email_notifications.py 
13
14
15
16
@step("Gather all tasks that recently failed")
def gather_failed_tasks() -> State:
    """Gather all tasks that have failed."""
    return {"failed_tasks": get_failed_tasks()}

send_email_notifications(state)

Send out an email notification for all tasks that have failed.

Source code in gso/workflows/tasks/send_email_notifications.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
@step("Send notification emails for all failed tasks")
def send_email_notifications(state: State) -> None:
    """Send out an email notification for all tasks that have failed."""
    general_settings = load_oss_params().GENERAL
    all_alerts = ""
    for failure in state["failed_tasks"]:
        failed_task_url = f"{general_settings.public_hostname}/workflows/{failure["process_id"]}"
        failed_subscription = get_subscription_by_process_id(failure["process_id"])
        all_alerts = f"{all_alerts}------\n\n"
        if failed_subscription:
            all_alerts = (
                f"{all_alerts}Product name: {failed_subscription.product.name}\n"
                f"Description: {failed_subscription.description}\n"
            )
        all_alerts = (
            f'{all_alerts}The step "{failure["last_step"]}" failed for the following reason: '
            f'"{failure["failed_reason"]}".\n\nPlease inspect the full workflow at the following link: '
            f'{failed_task_url}.\n\n'
        )

    send_mail(
        f"GAP {general_settings.environment} environment - One or more tasks have failed!",
        (
            f"Please check the following tasks in GAP which have failed.\n\n{all_alerts}------"
            f"\n\nRegards, the GÉANT Automation Platform.\n\n"
        ),
    )

task_send_email_notifications()

Gather all failed tasks, and send an email notification if needed.

Source code in gso/workflows/tasks/send_email_notifications.py 
48
49
50
51
52
53
@workflow("Send email notifications for all failed tasks", target=Target.SYSTEM)
def task_send_email_notifications() -> StepList:
    """Gather all failed tasks, and send an email notification if needed."""
    tasks_have_failed = conditional(lambda state: len(state["failed_tasks"]) > 0)

    return init >> gather_failed_tasks >> tasks_have_failed(send_email_notifications) >> done