E-mail in case a task in an ADF pipeline fails.

adf_email_on_failure.py

#!/usr/bin/env python3
"""A simple script to generate ADF + HTML code for a pretty e-mail report in case of an activity failure.
Every e-mail + fail activity has to be separate. Per each pair of activities, uncomment one line in the
ACTIVITIES constant to generate an increasingly populated report.

E-mail subject should be:
@concat('The pipeline ', pipeline().pipelineName, ' has failed.')
"""

import html
import platform
import shutil
import string
import subprocess
from typing import Final

# Uncomment these activities one by one to get increasingly better table code per fail e-mail
ACTIVITIES: Final = (
    "silver_to_gold",
    # "data_quality_checks_silver_to_gold",
)

HEADER_TEMPLATE: Final = """<p>You are receiving this e-mail because your pipeline has failed. Please see the specifics below:</p>

<ul>
  <li>Run ID - @{pipeline().RunId}</li>
  <li>Pipeline Name - @{pipeline().PipelineName}</li>
</ul>
<table border="1" cellpadding="5" cellspacing="0" style="border-collapse: collapse; width: 100%;">
  <tr style="background-color: #f2f2f2;">
    <th align="left">Status</th>
    <th align="left">Activity run ID</th>
    <th align="left">Activity Name</th>
    <th align="left">Execution Time</th>
    <th align="left">Error Message</th>
  </tr>"""

ROW_TEMPLATE: Final = string.Template("""<tr>
<td style="@{if(equals(activity('${activity}').status, 'Failed'), 'background-color: #FFCCCB;', 'background-color: #90EE90;')}">@{activity('${activity}').status}</td>
<td>@{activity('${activity}').ActivityRunId}</td>
<td>${activity}</td>
<td>@{formatDateTime(activity('${activity}').ExecutionStartTime, 'yyyy-MM-dd HH:mm:ss')}</td>
<td>@{if(equals(activity('${activity}').status, 'Failed'), concat(activity('${activity}').output.result.error.ename, ':', '<br>', activity('${activity}').output.result.error.evalue), '')}</td>
</tr>""")


def copy_to_clipboard(text: str) -> None:
    """Copy text to the clipboard."""
    match os := platform.system():
        case "Windows":
            cmd = "clip"
        case "Darwin":
            cmd = "pbcopy"
        case "Linux" | "FreeBSD":
            for cmd in ("xclip", "xsel"):
                if shutil.which(cmd):
                    break
            else:
                raise NotImplementedError(
                    f"If your {os} machine does not use xclip or xsel, please "
                    "use 'pyperclip' PyPi to copy text to the clipboard."
                )
        case _:
            raise NotImplementedError(f"Operating system {os} is not supported.")

    subprocess.run(cmd, text=True, check=False, input=text)


def main() -> int:
    """Main function."""
    copy_to_clipboard(
        "\n".join(
            [HEADER_TEMPLATE]
            + [ROW_TEMPLATE.substitute(activity=html.escape(activity)) for activity in ACTIVITIES]
            + ["</table>"]
        )
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())