airflow/models/dag.py
def clear(
        self,
        task_ids=None,
        start_date=None,
        end_date=None,
        only_failed=False,
        only_running=False,
        confirm_prompt=False,
        include_subdags=True,
        include_parentdag=True,
        dag_run_state: DagRunState = DagRunState.QUEUED,
        dry_run=False,
        session=None,
        get_tis=False,
        recursion_depth=0,
        max_recursion_depth=None,
        dag_bag=None,
        exclude_task_ids: FrozenSet[str] = frozenset({}),
    ):
        """
        Clears a set of task instances associated with the current dag for
        a specified date range.

        :param task_ids: List of task ids to clear
        :type task_ids: List[str]
        :param start_date: The minimum execution_date to clear
        :type start_date: datetime.datetime or None
        :param end_date: The maximum execution_date to clear
        :type end_date: datetime.datetime or None
        :param only_failed: Only clear failed tasks
        :type only_failed: bool
        :param only_running: Only clear running tasks.
        :type only_running: bool
        :param confirm_prompt: Ask for confirmation
        :type confirm_prompt: bool
        :param include_subdags: Clear tasks in subdags and clear external tasks
            indicated by ExternalTaskMarker
        :type include_subdags: bool
        :param include_parentdag: Clear tasks in the parent dag of the subdag.
        :type include_parentdag: bool
        :param dag_run_state: state to set DagRun to. If set to False, dagrun state will not
            be changed.
        :param dry_run: Find the tasks to clear but don't clear them.
        :type dry_run: bool
        :param session: The sqlalchemy session to use
        :type session: sqlalchemy.orm.session.Session
        :param dag_bag: The DagBag used to find the dags subdags (Optional)
        :type dag_bag: airflow.models.dagbag.DagBag
        :param exclude_task_ids: A set of ``task_id`` that should not be cleared
        :type exclude_task_ids: frozenset
        """
        if get_tis:
            warnings.warn(
                "Passing `get_tis` to dag.clear() is deprecated. Use `dry_run` parameter instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            dry_run = True

        if recursion_depth:
            warnings.warn(
                "Passing `recursion_depth` to dag.clear() is deprecated.",
                DeprecationWarning,
                stacklevel=2,
            )
        if max_recursion_depth:
            warnings.warn(
                "Passing `max_recursion_depth` to dag.clear() is deprecated.",
                DeprecationWarning,
                stacklevel=2,
            )

        state = []
        if only_failed:
            state += [State.FAILED, State.UPSTREAM_FAILED]
            only_failed = None
        if only_running:
            # Yes, having `+=` doesn't make sense, but this was the existing behaviour
            state += [State.RUNNING]
            only_running = None

        tis = self._get_task_instances(
            task_ids=task_ids,
            start_date=start_date,
            end_date=end_date,
            run_id=None,
            state=state,
            include_subdags=include_subdags,
            include_parentdag=include_parentdag,
            include_dependent_dags=include_subdags,  # compat, yes this is not a typo
            as_pk_tuple=False,
            session=session,
            dag_bag=dag_bag,
            exclude_task_ids=exclude_task_ids,
        )

        if dry_run:
            return tis

        tis = tis.all()

        count = len(tis)
        do_it = True
        if count == 0:
            return 0
        if confirm_prompt:
            ti_list = "\n".join(str(t) for t in tis)
            question = (
                "You are about to delete these {count} tasks:\n{ti_list}\n\nAre you sure? (yes/no): "
            ).format(count=count, ti_list=ti_list)
            do_it = utils.helpers.ask_yesno(question)

        if do_it:
            clear_task_instances(
                tis,
                session,
                dag=self,
                dag_run_state=dag_run_state,
            )
        else:
            count = 0
            print("Cancelled, nothing was cleared.")

        session.flush()
        return count

        Clears a set of task instances associated with the current dag for
        a specified date range.

        :param task_ids: List of task ids to clear
        :type task_ids: List[str]
        :param start_date: The minimum execution_date to clear
        :type start_date: datetime.datetime or None
        :param end_date: The maximum execution_date to clear
        :type end_date: datetime.datetime or None
        :param only_failed: Only clear failed tasks
        :type only_failed: bool
        :param only_running: Only clear running tasks.
        :type only_running: bool
        :param confirm_prompt: Ask for confirmation
        :type confirm_prompt: bool
        :param include_subdags: Clear tasks in subdags and clear external tasks
            indicated by ExternalTaskMarker
        :type include_subdags: bool
        :param include_parentdag: Clear tasks in the parent dag of the subdag.
        :type include_parentdag: bool
        :param dag_run_state: state to set DagRun to. If set to False, dagrun state will not
            be changed.
        :param dry_run: Find the tasks to clear but don't clear them.
        :type dry_run: bool
        :param session: The sqlalchemy session to use
        :type session: sqlalchemy.orm.session.Session
        :param dag_bag: The DagBag used to find the dags subdags (Optional)
        :type dag_bag: airflow.models.dagbag.DagBag
        :param exclude_task_ids: A set of ``task_id`` that should not be cleared
        :type exclude_task_ids: frozenset
        
Clears a set of task instances associated with the current dag for
a specified date range.
Clears a set of task instances associated with the current dag for a specified date range .
def clear(
        self,
        task_ids=None,
        start_date=None,
        end_date=None,
        only_failed=False,
        only_running=False,
        confirm_prompt=False,
        include_subdags=True,
        include_parentdag=True,
        dag_run_state: DagRunState = DagRunState.QUEUED,
        dry_run=False,
        session=None,
        get_tis=False,
        recursion_depth=0,
        max_recursion_depth=None,
        dag_bag=None,
        exclude_task_ids: FrozenSet[str] = frozenset({}),
    ):
        if get_tis:
            warnings.warn(
                "Passing `get_tis` to dag.clear() is deprecated. Use `dry_run` parameter instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            dry_run = True
        if recursion_depth:
            warnings.warn(
                "Passing `recursion_depth` to dag.clear() is deprecated.",
                DeprecationWarning,
                stacklevel=2,
            )
        if max_recursion_depth:
            warnings.warn(
                "Passing `max_recursion_depth` to dag.clear() is deprecated.",
                DeprecationWarning,
                stacklevel=2,
            )
        state = []
        if only_failed:
            state += [State.FAILED, State.UPSTREAM_FAILED]
            only_failed = None
        if only_running:
            state += [State.RUNNING]
            only_running = None
        tis = self._get_task_instances(
            task_ids=task_ids,
            start_date=start_date,
            end_date=end_date,
            run_id=None,
            state=state,
            include_subdags=include_subdags,
            include_parentdag=include_parentdag,
            include_dependent_dags=include_subdags,  
            as_pk_tuple=False,
            session=session,
            dag_bag=dag_bag,
            exclude_task_ids=exclude_task_ids,
        )
        if dry_run:
            return tis
        tis = tis.all()
        count = len(tis)
        do_it = True
        if count == 0:
            return 0
        if confirm_prompt:
            ti_list = "\n".join(str(t) for t in tis)
            question = (
                "You are about to delete these {count} tasks:\n{ti_list}\n\nAre you sure? (yes/no): "
            ).format(count=count, ti_list=ti_list)
            do_it = utils.helpers.ask_yesno(question)
        if do_it:
            clear_task_instances(
                tis,
                session,
                dag=self,
                dag_run_state=dag_run_state,
            )
        else:
            count = 0
            print("Cancelled, nothing was cleared.")
        session.flush()
        return count
def clear ( self , task_ids = None , start_date = None , end_date = None , only_failed = False , only_running = False , confirm_prompt = False , include_subdags = True , include_parentdag = True , dag_run_state : DagRunState = DagRunState . QUEUED , dry_run = False , session = None , get_tis = False , recursion_depth = 0 , max_recursion_depth = None , dag_bag = None , exclude_task_ids : FrozenSet [ str ] = frozenset ( { } ) , ) : if get_tis : warnings . warn ( "Passing `get_tis` to dag.clear() is deprecated. Use `dry_run` parameter instead." , DeprecationWarning , stacklevel = 2 , ) dry_run = True if recursion_depth : warnings . warn ( "Passing `recursion_depth` to dag.clear() is deprecated." , DeprecationWarning , stacklevel = 2 , ) if max_recursion_depth : warnings . warn ( "Passing `max_recursion_depth` to dag.clear() is deprecated." , DeprecationWarning , stacklevel = 2 , ) state = [ ] if only_failed : state += [ State . FAILED , State . UPSTREAM_FAILED ] only_failed = None if only_running : state += [ State . RUNNING ] only_running = None tis = self . _get_task_instances ( task_ids = task_ids , start_date = start_date , end_date = end_date , run_id = None , state = state , include_subdags = include_subdags , include_parentdag = include_parentdag , include_dependent_dags = include_subdags , as_pk_tuple = False , session = session , dag_bag = dag_bag , exclude_task_ids = exclude_task_ids , ) if dry_run : return tis tis = tis . all ( ) count = len ( tis ) do_it = True if count == 0 : return 0 if confirm_prompt : ti_list = "\n" . join ( str ( t ) for t in tis ) question = ( "You are about to delete these {count} tasks:\n{ti_list}\n\nAre you sure? (yes/no): " ) . format ( count = count , ti_list = ti_list ) do_it = utils . helpers . ask_yesno ( question ) if do_it : clear_task_instances ( tis , session , dag = self , dag_run_state = dag_run_state , ) else : count = 0 print ( "Cancelled, nothing was cleared." ) session . flush ( ) return count
Clears a set of task instances associated with the current dag for
a specified date range.
Clears a set of task instances associated with the current dag for a specified date range .
"""
        Clears a set of task instances associated with the current dag for
        a specified date range.

        :param task_ids: List of task ids to clear
        :type task_ids: List[str]
        :param start_date: The minimum execution_date to clear
        :type start_date: datetime.datetime or None
        :param end_date: The maximum execution_date to clear
        :type end_date: datetime.datetime or None
        :param only_failed: Only clear failed tasks
        :type only_failed: bool
        :param only_running: Only clear running tasks.
        :type only_running: bool
        :param confirm_prompt: Ask for confirmation
        :type confirm_prompt: bool
        :param include_subdags: Clear tasks in subdags and clear external tasks
            indicated by ExternalTaskMarker
        :type include_subdags: bool
        :param include_parentdag: Clear tasks in the parent dag of the subdag.
        :type include_parentdag: bool
        :param dag_run_state: state to set DagRun to. If set to False, dagrun state will not
            be changed.
        :param dry_run: Find the tasks to clear but don't clear them.
        :type dry_run: bool
        :param session: The sqlalchemy session to use
        :type session: sqlalchemy.orm.session.Session
        :param dag_bag: The DagBag used to find the dags subdags (Optional)
        :type dag_bag: airflow.models.dagbag.DagBag
        :param exclude_task_ids: A set of ``task_id`` that should not be cleared
        :type exclude_task_ids: frozenset
        """ # Yes, having `+=` doesn't make sense, but this was the existing behaviour # compat, yes this is not a typo
self
task_ids
List of task ids to clear
List of task ids to clear
start_date
The minimum execution_date to clear
The minimum execution_date to clear
end_date
The maximum execution_date to clear
The maximum execution_date to clear
only_failed
Only clear failed tasks
Only clear failed tasks
only_running
Only clear running tasks.
Only clear running tasks .
confirm_prompt
Ask for confirmation
Ask for confirmation
include_subdags
Clear tasks in subdags and clear external tasks
indicated by ExternalTaskMarker
Clear tasks in subdags and clear external tasks indicated by ExternalTaskMarker
include_parentdag
Clear tasks in the parent dag of the subdag.
Clear tasks in the parent dag of the subdag .
dag_run_state
DagRunState
state to set DagRun to. If set to False, dagrun state will not
be changed.
state to set DagRun to . If set to False dagrun state will not be changed .
dry_run
Find the tasks to clear but don't clear them.
Find the tasks to clear but don ' t clear them .
session
The sqlalchemy session to use
The sqlalchemy session to use
get_tis
recursion_depth
max_recursion_depth
dag_bag
The DagBag used to find the dags subdags (Optional)
The DagBag used to find the dags subdags ( Optional )
exclude_task_ids
FrozenSet[str]
A set of ``task_id`` that should not be cleared
A set of ` ` task_id ` ` that should not be cleared