发送有关 Luigi 作业失败的松弛消息答案

【问题标题】：Send a slack message on Luigi job failure发送有关 Luigi 作业失败的松弛消息
【发布时间】：2020-03-10 15:04:19
【问题描述】：

让任何Luigi 任务失败向 Slack 发布消息的最佳方式是什么？我知道任务有一个可覆盖的 on_failure 方法，但是有没有办法指定全局策略而不是每个任务？

我目前的方法是我有一个覆盖 on_failure 方法的 SlackOnFailureMixin（粘贴在下面），但我必须手动将它混合到我想要拥有此功能的每个 luigi.Task 中，如果我忘记了，那么不发布失败。

这是我的 SlackOnFailure mixin 的来源，以防有人对这种方法感到好奇。

class SlackOnFailureMixin:
    """
    IMPORTANT NOTE! Make sure this mixin goes first in your class list, i.e.

        class MyTask(SlackOnFailureMixin, luigi.Task):

    Due to Python's method resolution order (MRO), this won't work unless you
    mix it in first.
    """

    def on_failure(self, exception):
        message = super().on_failure(exception)
        if "SLACK_ON_FAILURE" in os.environ:
            slack.eng_errors.post_error_message(
                f"Error in Luigi Task: {self.task_id}", message
            )
        return message

谁能推荐一个更好的方法来做到这一点？谢谢！

【问题讨论】：

标签： luigi

【解决方案1】：

你总是可以创建一个你继承的基类，将 Luigi 扩展到你需要的东西。

class BaseTask(luigi.Task):
    def on_failure(self, exception):
        message = super().on_failure(exception)
        if "SLACK_ON_FAILURE" in os.environ:
            slack.eng_errors.post_error_message(
                f"Error in Luigi Task: {self.task_id}", message
            )
        return message

这几乎是唯一（也是最好）的方法。您实际上不需要更改任何 mixin 的代码，只需将其放入基类中即可。

【讨论】：