https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
try:
            
            with create_session() as session:
                pickle = DagPickle(dag)
                session.add(pickle)
            pickle_id = pickle.id
            
            print(f"Pickled dag {dag} as pickle_id: {pickle_id}")
        except Exception as e:
            print("Could not pickle the DAG")
            print(e)
            raise e
    executor = ExecutorLoader.get_default_executor()
    executor.job_id = None
    executor.start()
    print("Sending to executor.")
    executor.queue_task_instance(
        ti,
        mark_success=args.mark_success,
        pickle_id=pickle_id,
        ignore_all_deps=args.ignore_all_dependencies,
        ignore_depends_on_past=should_ignore_depends_on_past(args),
        wait_for_past_depends_before_skipping=(args.depends_on_past == "wait"),
        ignore_task_deps=args.ignore_dependencies,
        ignore_ti_state=args.force,
        pool=args.pool,
    )
    executor.heartbeat()
    executor.end()
def _run_task_by_local_task_job(args, ti: TaskInstance) -> TaskReturnCode | None: