airflow/api/common/experimental/get_task.py
def get_task(dag_id, task_id):
    """Return the task object identified by the given dag_id and task_id."""
    dagbag = DagBag()

    # Check DAG exists.
    if dag_id not in dagbag.dags:
        error_message = "Dag id {} not found".format(dag_id)
        raise DagNotFound(error_message)

    # Get DAG object and check Task Exists
    dag = dagbag.get_dag(dag_id)
    if not dag.has_task(task_id):
        error_message = 'Task {} not found in dag {}'.format(task_id, dag_id)
        raise TaskNotFound(error_message)

    # Return the task.
    return dag.get_task(task_id)
def get_task(dag_id, task_id):
    """Return the task object identified by the given dag_id and task_id."""
    dagbag = DagBag()

    # Check DAG exists.
    if dag_id not in dagbag.dags:
        error_message = "Dag id {} not found".format(dag_id)
        raise DagNotFound(error_message)

    # Get DAG object and check Task Exists
    dag = dagbag.get_dag(dag_id)
    if not dag.has_task(task_id):
        error_message = 'Task {} not found in dag {}'.format(task_id, dag_id)
        raise TaskNotFound(error_message)

    # Return the task.
    return dag.get_task(task_id)
def get_task ( dag_id , task_id ) : dagbag = DagBag ( ) # Check DAG exists. if dag_id not in dagbag . dags : error_message = "Dag id {} not found" . format ( dag_id ) raise DagNotFound ( error_message ) # Get DAG object and check Task Exists dag = dagbag . get_dag ( dag_id ) if not dag . has_task ( task_id ) : error_message = 'Task {} not found in dag {}' . format ( task_id , dag_id ) raise TaskNotFound ( error_message ) # Return the task. return dag . get_task ( task_id )
Return the task object identified by the given dag_id and task_id.
Return the task object identified by the given dag_id and task_id .