In my DAG file, I have define a on_failure_callback() function to post a Slack in case of failure.
It works well if I specify for each operator in my DAG : on_failure_callback=on_failure_callback()
Is there a way to automate (via default_args for instance, or via my DAG object) the dispatch to all of my operators?
Late answer here, but yes you can specify an on_failure_callback in defaults for the DAG. You just have to write a custom function, making sure it can take in the context. Example:
def failure_callback(context):
message = [
":red_circle: Task failed",
f"*Dag*: {context['dag_run'].dag_id}",
f"*Run*: {context['dag_run'].run_id}",
f"*Task*: <{context.get('task_instance').log_url}|*{context.get('task_instance').task_id}* failed for execution {context.get('execution_date')}>",
]
# Replace this return with whatever you want
# I usually send a Slack notification here
return "\n".join(message)
with DAG(
...
default_args={
...
"on_failure_callback": failure_callback,
},
) as dag:
...