I want to programmatically clear tasks in airflow based the number of times they have already been cleared. I need a way to store how many times tasks have been cleared, such that when the tasks are cleared, this stored value is not erased.
I tried using the params dictionary - as you can see below, if the dictionary is empty, I update it, then clear the tasks. However, this empties the params dict, and the dag gets stuck in a loop.
tasks_to_clear is going to be all tasks except 'first_step'.
What mechanism should I use to control clear_task? (keep in mind the below is not all of my code, just what I felt was necessary to include)
from airflow.models.taskinstance import clear_task_instances
def my_failure_function(context):
dag_run = context.get('dag_run')
task_instances = dag_run.get_task_instances()
tasks_to_clear = [t for t in task_instances if t.task_id not in ('first_step')]
@provide_session
def clear_task(task_ids, session = None, dag = None):
clear_task_instances(tis = task_ids, session = session, dag = dag)
if not context['params']:
context['params']['clear_task_count'] = True
print('printing params before clear_task')
print(context['params'])
clear_task(task_ids = tasks_to_clear, dag = dag)