Move DagBag to airflow/dag_processing

.pre-commit-config.yaml

@@ -1580,6 +1580,7 @@ repos:
           ^airflow-core/src/airflow/dag_processing/collection\.py$|
           ^airflow-core/src/airflow/dag_processing/manager\.py$|
           ^airflow-core/src/airflow/dag_processing/processor\.py$|
+          ^airflow-core/src/airflow/dag_processing/dagbag\.py$|
           ^airflow-core/src/airflow/datasets/metadata\.py$|
           ^airflow-core/src/airflow/exceptions\.py$|
           ^airflow-core/src/airflow/executors/local_executor\.py$|
@@ -1593,7 +1594,6 @@ repos:
           ^airflow-core/src/airflow/models/baseoperator\.py$|
           ^airflow-core/src/airflow/models/connection\.py$|
           ^airflow-core/src/airflow/models/dag\.py$|
-          ^airflow-core/src/airflow/models/dagbag\.py$|
           ^airflow-core/src/airflow/models/dagrun\.py$|
           ^airflow-core/src/airflow/models/deadline\.py$|
           ^airflow-core/src/airflow/models/expandinput\.py$|

airflow-core/docs/best-practices.rst

@@ -727,7 +727,7 @@ Unit tests ensure that there is no incorrect code in your Dag. You can write uni
 
     import pytest
 
-    from airflow.models import DagBag
+    from airflow.dag_processing.dagbag import DagBag
 
 
     @pytest.fixture()

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_report.py

@@ -34,7 +34,7 @@
     ReadableDagsFilterDep,
     requires_access_dag,
 )
-from airflow.models.dagbag import DagBag
+from airflow.dag_processing.dagbag import DagBag
 
 dag_report_router = AirflowRouter(tags=["DagReport"], prefix="/dagReports")
 

airflow-core/src/airflow/cli/commands/dag_command.py

@@ -37,11 +37,11 @@
 from airflow.cli.simple_table import AirflowConsole
 from airflow.cli.utils import fetch_dag_run_from_run_id_or_logical_date_string
 from airflow.dag_processing.bundles.manager import DagBundlesManager
+from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db
 from airflow.exceptions import AirflowConfigException, AirflowException
 from airflow.jobs.job import Job
-from airflow.models import DagBag, DagModel, DagRun, TaskInstance
+from airflow.models import DagModel, DagRun, TaskInstance
 from airflow.models.dag import get_next_data_interval
-from airflow.models.dagbag import sync_bag_to_db
 from airflow.models.errors import ParseImportError
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.utils import cli as cli_utils
@@ -364,7 +364,7 @@ def dag_list_dags(args, session: Session = NEW_SESSION) -> None:
     dagbag_import_errors = 0
     dags_list = []
     if args.local:
-        from airflow.models.dagbag import DagBag
+        from airflow.dag_processing.dagbag import DagBag
 
         # Get import errors from the local area
         if args.bundle_name:
@@ -463,6 +463,7 @@ def dag_list_import_errors(args, session: Session = NEW_SESSION) -> None:
 
     if args.local:
         # Get import errors from local areas
+
         if args.bundle_name:
             manager = DagBundlesManager()
             validate_dag_bundle_arg(args.bundle_name)