diff --git a/airflow/executors/executor_loader.py b/airflow/executors/executor_loader.py
index 1eeee1ff68a9..ec79860918b8 100644
--- a/airflow/executors/executor_loader.py
+++ b/airflow/executors/executor_loader.py
@@ -173,6 +173,8 @@ def set_default_executor(cls, executor: BaseExecutor) -> None:
This is used in rare cases such as dag.run which allows, as a user convenience, to provide
the executor by cli/argument instead of Airflow configuration
+
+ todo: given comments above, is this needed anymore since DAG.run is removed?
"""
exec_class_name = executor.__class__.__qualname__
exec_name = ExecutorName(f"{executor.__module__}.{exec_class_name}")
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 2dc425daa054..5cc5cf443140 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -28,7 +28,6 @@
import sys
import time
import traceback
-import warnings
import weakref
from collections import abc, defaultdict, deque
from contextlib import ExitStack
@@ -88,13 +87,11 @@
DuplicateTaskIdFound,
FailStopDagInvalidTriggerRule,
ParamValidationError,
- RemovedInAirflow3Warning,
TaskDeferred,
TaskNotFound,
UnknownExecutorException,
)
from airflow.executors.executor_loader import ExecutorLoader
-from airflow.jobs.job import run_job
from airflow.models.abstractoperator import AbstractOperator, TaskStateChangeCallback
from airflow.models.asset import (
AssetDagRunQueue,
@@ -2296,84 +2293,8 @@ def _remove_task(self, task_id: str) -> None:
self.task_count = len(self.task_dict)
- def run(
- self,
- start_date=None,
- end_date=None,
- mark_success=False,
- local=False,
- donot_pickle=airflow_conf.getboolean("core", "donot_pickle"),
- ignore_task_deps=False,
- ignore_first_depends_on_past=True,
- pool=None,
- delay_on_limit_secs=1.0,
- verbose=False,
- conf=None,
- rerun_failed_tasks=False,
- run_backwards=False,
- run_at_least_once=False,
- continue_on_failures=False,
- disable_retry=False,
- ):
- """
- Run the DAG.
-
- :param start_date: the start date of the range to run
- :param end_date: the end date of the range to run
- :param mark_success: True to mark jobs as succeeded without running them
- :param local: True to run the tasks using the LocalExecutor
- :param donot_pickle: True to avoid pickling DAG object and send to workers
- :param ignore_task_deps: True to skip upstream tasks
- :param ignore_first_depends_on_past: True to ignore depends_on_past
- dependencies for the first set of tasks only
- :param pool: Resource pool to use
- :param delay_on_limit_secs: Time in seconds to wait before next attempt to run
- dag run when max_active_runs limit has been reached
- :param verbose: Make logging output more verbose
- :param conf: user defined dictionary passed from CLI
- :param rerun_failed_tasks:
- :param run_backwards:
- :param run_at_least_once: If true, always run the DAG at least once even
- if no logical run exists within the time range.
- """
- warnings.warn(
- "`DAG.run()` is deprecated and will be removed in Airflow 3.0. Consider "
- "using `DAG.test()` instead, or trigger your dag via API.",
- RemovedInAirflow3Warning,
- stacklevel=2,
- )
-
- from airflow.executors.executor_loader import ExecutorLoader
- from airflow.jobs.backfill_job_runner import BackfillJobRunner
-
- if local:
- from airflow.executors.local_executor import LocalExecutor
-
- ExecutorLoader.set_default_executor(LocalExecutor())
-
- from airflow.jobs.job import Job
-
- job = Job()
- job_runner = BackfillJobRunner(
- job=job,
- dag=self,
- start_date=start_date,
- end_date=end_date,
- mark_success=mark_success,
- donot_pickle=donot_pickle,
- ignore_task_deps=ignore_task_deps,
- ignore_first_depends_on_past=ignore_first_depends_on_past,
- pool=pool,
- delay_on_limit_secs=delay_on_limit_secs,
- verbose=verbose,
- conf=conf,
- rerun_failed_tasks=rerun_failed_tasks,
- run_backwards=run_backwards,
- run_at_least_once=run_at_least_once,
- continue_on_failures=continue_on_failures,
- disable_retry=disable_retry,
- )
- run_job(job=job, execute_callable=job_runner._execute)
+ def run(self, *args, **kwargs):
+ """Leaving this here to be removed in other PR for simpler review."""
def cli(self):
"""Exposes a CLI specific to this DAG."""
diff --git a/contributing-docs/quick-start-ide/contributors_quick_start_pycharm.rst b/contributing-docs/quick-start-ide/contributors_quick_start_pycharm.rst
index d830496b2720..4a3319ae97dd 100644
--- a/contributing-docs/quick-start-ide/contributors_quick_start_pycharm.rst
+++ b/contributing-docs/quick-start-ide/contributors_quick_start_pycharm.rst
@@ -78,35 +78,14 @@ It requires "airflow-env" virtual environment configured locally.
- Copy any example DAG present in the ``/airflow/example_dags`` directory to ``/files/dags/``.
-- Add a ``__main__`` block at the end of your DAG file to make it runnable. It will run a ``back_fill`` job:
+- Add a ``__main__`` block at the end of your DAG file to make it runnable:
.. code-block:: python
if __name__ == "__main__":
- dag.clear()
- dag.run()
+ dag.test()
-- Add ``AIRFLOW__CORE__EXECUTOR=DebugExecutor`` to Environment variable of Run Configuration.
-
- - Click on Add configuration
-
- .. raw:: html
-
-
-
-
-
-