Replace external_trigger check with DagRunType #45932
Description
Now that we have explicit Dag Run types, we should not need DagRun.external_trigger too.
Example, the following code can be changed
to
def choose_branch(self, context: Context) -> str | Iterable[str]:
# If the DAG Run is manually triggered, then return without
# skipping downstream tasks
dag_run: DagRun = context["dag_run"] # type: ignore[assignment]
if dag_run.run_type == DagRunType.MANUAL:
self.log.info("Externally triggered DAG_Run: allowing execution to proceed.")
return list(context["task"].get_direct_relative_ids(upstream=False))We should replace all usages of DagRun.external_trigger and remove it from DB too.