airflow/api/common/experimental/mark_tasks.py
def set_dag_run_state_to_failed(dag, execution_date, commit=False, session=None):
    """
    Set the dag run for a specific execution date and its running task instances
    to failed.
    :param dag: the DAG of which to alter state
    :param execution_date: the execution date from which to start looking
    :param commit: commit DAG and tasks to be altered to the database
    :param session: database session
    :return: If commit is true, list of tasks that have been updated,
             otherwise list of tasks that will be updated
    :raises: AssertionError if dag or execution_date is invalid
    """
    res = []

    if not dag or not execution_date:
        return res

    # Mark the dag run to failed.
    if commit:
        _set_dag_run_state(dag.dag_id, execution_date, State.FAILED, session)

    # Mark only RUNNING task instances.
    TI = TaskInstance
    task_ids = [task.task_id for task in dag.tasks]
    tis = session.query(TI).filter(
        TI.dag_id == dag.dag_id,
        TI.execution_date == execution_date,
        TI.task_id.in_(task_ids)).filter(TI.state == State.RUNNING)
    task_ids_of_running_tis = [ti.task_id for ti in tis]
    for task in dag.tasks:
        if task.task_id not in task_ids_of_running_tis:
            continue
        task.dag = dag
        new_state = set_state(task=task, execution_date=execution_date,
                              state=State.FAILED, commit=commit)
        res.extend(new_state)

    return res
def set_dag_run_state_to_failed(dag, execution_date, commit=False, session=None):
    """
    Set the dag run for a specific execution date and its running task instances
    to failed.
    :param dag: the DAG of which to alter state
    :param execution_date: the execution date from which to start looking
    :param commit: commit DAG and tasks to be altered to the database
    :param session: database session
    :return: If commit is true, list of tasks that have been updated,
             otherwise list of tasks that will be updated
    :raises: AssertionError if dag or execution_date is invalid
    """
    res = []

    if not dag or not execution_date:
        return res

    # Mark the dag run to failed.
    if commit:
        _set_dag_run_state(dag.dag_id, execution_date, State.FAILED, session)

    # Mark only RUNNING task instances.
    TI = TaskInstance
    task_ids = [task.task_id for task in dag.tasks]
    tis = session.query(TI).filter(
        TI.dag_id == dag.dag_id,
        TI.execution_date == execution_date,
        TI.task_id.in_(task_ids)).filter(TI.state == State.RUNNING)
    task_ids_of_running_tis = [ti.task_id for ti in tis]
    for task in dag.tasks:
        if task.task_id not in task_ids_of_running_tis:
            continue
        task.dag = dag
        new_state = set_state(task=task, execution_date=execution_date,
                              state=State.FAILED, commit=commit)
        res.extend(new_state)

    return res
def set_dag_run_state_to_failed ( dag , execution_date , commit = False , session = None ) : res = [ ] if not dag or not execution_date : return res # Mark the dag run to failed. if commit : _set_dag_run_state ( dag . dag_id , execution_date , State . FAILED , session ) # Mark only RUNNING task instances. TI = TaskInstance task_ids = [ task . task_id for task in dag . tasks ] tis = session . query ( TI ) . filter ( TI . dag_id == dag . dag_id , TI . execution_date == execution_date , TI . task_id . in_ ( task_ids ) ) . filter ( TI . state == State . RUNNING ) task_ids_of_running_tis = [ ti . task_id for ti in tis ] for task in dag . tasks : if task . task_id not in task_ids_of_running_tis : continue task . dag = dag new_state = set_state ( task = task , execution_date = execution_date , state = State . FAILED , commit = commit ) res . extend ( new_state ) return res
Set the dag run for a specific execution date and its running task instances
    to failed.
    :param dag: the DAG of which to alter state
    :param execution_date: the execution date from which to start looking
    :param commit: commit DAG and tasks to be altered to the database
    :param session: database session
    :return: If commit is true, list of tasks that have been updated,
             otherwise list of tasks that will be updated
    :raises: AssertionError if dag or execution_date is invalid
Set the dag run for a specific execution date and its running task instances to failed . : param dag : the DAG of which to alter state : param execution_date : the execution date from which to start looking : param commit : commit DAG and tasks to be altered to the database : param session : database session : return : If commit is true list of tasks that have been updated otherwise list of tasks that will be updated : raises : AssertionError if dag or execution_date is invalid