Каков наилучший способ, чтобы при любой ошибке Luigi опубликовать сообщение, чтобы расслабиться? Я знаю, что у задач есть переопределяемый метод on_failure, но есть ли способ указать глобальную политику вместо каждой задачи?
Мой текущий подход заключается в том, что у меня есть SlackOnFailureMixin (вставленный ниже), который переопределяет метод on_failure, но я должен вручную смешать его с каждым luigi.Task
, в котором я хочу включить эту функцию, и если я забуду, сбои не будут опубликованы.
Это источник моего миксина SlackOnFailure на случай, если кому-нибудь любопытно, что такое подход.
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
Кто-нибудь может порекомендовать лучший способ сделать это? Спасибо!