check_this_branch_none__t3__20211203

I AM YOUR FAILURE CALLBACK AHAHA
dict_keys(['conf', 'dag', 'dag_run', 'data_interval_end', 'data_interval_start', 'ds', 'ds_nodash', 'execution_date', 'inlets', 'logical_date', 'macros', 'next_ds', 'next_ds_nodash', 'next_execution_date', 'outlets', 'params', 'prev_data_interval_start_success', 'prev_data_interval_end_success', 'prev_ds', 'prev_ds_nodash', 'prev_execution_date', 'prev_execution_date_success', 'prev_start_date_success', 'run_id', 'task', 'task_instance', 'task_instance_key_str', 'test_mode', 'ti', 'tomorrow_ds', 'tomorrow_ds_nodash', 'ts', 'ts_nodash', 'ts_nodash_with_tz', 'var', 'conn', 'yesterday_ds', 'yesterday_ds_nodash', 'reason'])
KeysView(<airflow.configuration.AirflowConfigParser object at 0x7f4bbedd2e50>)
task_failure
branch_none__t3__20211203
<TaskInstance: branch_none.t3 manual__2021-12-03T10:00:34.643888+00:00 [upstream_failed]>
None

example-dag.py

from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.operators.dummy import DummyOperator
from airflow.operators.python_operator import PythonOperator


def fail(context):
    key=context["task_instance_key_str"]
    with open(f"/tmp/check_this_{key}", "w") as fp:
        print("I AM YOUR FAILURE CALLBACK AHAHA", file=fp)
        print(context.keys(), file=fp)
        print(context['conf'].keys(), file=fp)
        print(context['reason'], file=fp)
        print(context["task_instance_key_str"], file=fp)
        print(context["ti"], file=fp)
        print(context["ti"].error(), file=fp)

def foo(**context):
    return 1 / 0


with DAG(
    dag_id="branch_none",
    schedule_interval=None,
    start_date=days_ago(3),
    catchup=False,
    on_failure_callback=fail,
) as dag:
    op1 = PythonOperator(task_id="br", python_callable=foo)
    op2 = DummyOperator(task_id="t3")

    op1 >> op2