https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
task = _dag.get_task(task_id=args.task_id)
    ti, _ = _get_ti(task, args.map_index, exec_date_or_run_id=args.execution_date_or_run_id, pool=args.pool)
    ti.init_run_context(raw=args.raw)
    hostname = get_hostname()
    log.info("Running %s on host %s", ti, hostname)
    
    
    
    
    
    
    settings.reconfigure_orm(disable_connection_pool=True)
    task_return_code = None
    try:
        if args.interactive:
            task_return_code = _run_task_by_selected_method(args, _dag, ti)
        else:
            with _move_task_handlers_to_root(ti), _redirect_stdout_to_ti_log(ti):
                task_return_code = _run_task_by_selected_method(args, _dag, ti)
                if task_return_code == TaskReturnCode.DEFERRED:
                    _set_task_deferred_context_var()
    finally:
        try:
            get_listener_manager().hook.before_stopping(component=TaskCommandMarker())
        except Exception:
            pass
    return task_return_code
@cli_utils.action_cli(check_db=False)
@providers_configuration_loaded
def task_failed_deps(args) -> None: