def clear_task_instances(tis, session, activate_dag_runs=True):
 job_ids = []
 for ti in tis:
  if (ti.state == State.RUNNING):
   if ti.job_id:
    ti.state = State.SHUTDOWN
    job_ids.append(ti.job_id)
  else:
   session.delete(ti)
 if job_ids:
  from airflow.jobs import BaseJob as BJ
  for job in session.query(BJ).filter(BJ.id.in_(job_ids)).all():
   job.state = State.SHUTDOWN
 if activate_dag_runs:
  execution_dates = {ti.execution_date for ti in tis}
  dag_ids = {ti.dag_id for ti in tis}
  drs = session.query(DagRun).filter(DagRun.dag_id.in_(dag_ids), DagRun.execution_date.in_(execution_dates)).all()
  for dr in drs:
   dr.state = State.RUNNING
   dr.start_date = datetime.now()

def clear_task_instances ( tis , session , activate_dag_runs = True ) : job_ids = [ ] for ti in tis : if ( ti . state == State . RUNNING ) : if ti . job_id : ti . state = State . SHUTDOWN job_ids . append ( ti . job_id ) else : session . delete ( ti ) if job_ids : from airflow . jobs import BaseJob as BJ for job in session . query ( BJ ) . filter ( BJ . id . in_ ( job_ids ) ) . all ( ) : job . state = State . SHUTDOWN if activate_dag_runs : execution_dates = { ti . execution_date for ti in tis } dag_ids = { ti . dag_id for ti in tis } drs = session . query ( DagRun ) . filter ( DagRun . dag_id . in_ ( dag_ids ) , DagRun . execution_date . in_ ( execution_dates ) ) . all ( ) for dr in drs : dr . state = State . RUNNING dr . start_date = datetime . now ( )
clears a set of task instances .