python - 如何在 sla_miss_callback 函数中获取上下文对象

Question

我能够成功实现和测试 on_success_callback 和 on_failure_callback 在 Apache Airflow 中，包括能够使用上下文对象成功地将参数传递给它们。但是我无法成功实现 sla_miss_callback .通过浏览不同的在线资源，我发现传递给这个函数的参数是

dag, task_list, blocking_task_list, slas, blocking_tis

然而 sla_miss_callback 与成功/失败回调不同，它不会在其参数列表中获取上下文对象，如果我尝试运行多组运算符(如 Python、Bash 运算符)，它们会失败，并且调度程序会提示未传递上下文来执行函数。

我尝试查看其他在线资源，仅在其中一个 ( https://www.rea-group.com/blog/watching-the-watcher/) 中我发现我们可以通过使用 self 对象来提取上下文对象。所以我将 self 附加到上述额外的 5 个参数，但它对我不起作用。我想知道如何将上下文对象检索或传递给 sla_miss_callback 函数不仅用于运行不同的操作符，还用于检索有关未满足 SLA 的 dag 的其他详细信息

Answer 1

似乎不可能将上下文字典传递给 SLA 回调(请参阅 sla_miss_callback 的源代码)，但我找到了一种合理的解决方法来访问有关 dag-run 的其他一些信息，例如 dag_id、task_id 和 execution_date .您还可以使用任何应该可以正常工作的内置宏/参数。当我使用 SlackWebhookOperator对于我的其他回调，我使用 SlackWebhookHook对于 sla_miss_callback。例如:

from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook

def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis, *args, **kwargs):
dag_id = slas[0].dag_id
task_id = slas[0].task_id
execution_date = slas[0].execution_date.isoformat()
hook = SlackWebhookHook(
    http_conn_id='slack_callbacks',
    webhook_token=BaseHook.get_connection('slack_callbacks').password,
    message=f"""
        :sos: *SLA has been missed*
        *Task:* {task_id}
        *DAG:* {dag_id}
        *Execution Date:* {execution_date}
        """
)
hook.execute()