Revert "AIP-66: Add support for parsing DAG bundles (#45371)"

This reverts commit 72ab1d5.

Author: jedcunningham

airflow/cli/commands/local_commands/dag_processor_command.py

@@ -39,6 +39,7 @@ def _create_dag_processor_job_runner(args: Any) -> DagProcessorJobRunner:
         job=Job(),
         processor=DagFileProcessorManager(
             processor_timeout=processor_timeout_seconds,
+            dag_directory=args.subdir,
             max_runs=args.num_runs,
         ),
     )

airflow/dag_processing/bundles/manager.py

@@ -64,23 +64,6 @@ def parse_config(self) -> None:
                 "Bundle config is not a list. Check config value"
                 " for section `dag_bundles` and key `backends`."
             )
-
-        # example dags!
-        if conf.getboolean("core", "LOAD_EXAMPLES"):
-            from airflow import example_dags
-
-            example_dag_folder = next(iter(example_dags.__path__))
-            backends.append(
-                {
-                    "name": "example_dags",
-                    "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                    "kwargs": {
-                        "local_folder": example_dag_folder,
-                        "refresh_interval": conf.getint("scheduler", "dag_dir_list_interval"),
-                    },
-                }
-            )
-
         seen = set()
         for cfg in backends:
             name = cfg["name"]

airflow/dag_processing/collection.py

@@ -74,14 +74,11 @@
 log = logging.getLogger(__name__)
 
 
-def _create_orm_dags(
-    bundle_name: str, dags: Iterable[MaybeSerializedDAG], *, session: Session
-) -> Iterator[DagModel]:
+def _create_orm_dags(dags: Iterable[MaybeSerializedDAG], *, session: Session) -> Iterator[DagModel]:
     for dag in dags:
         orm_dag = DagModel(dag_id=dag.dag_id)
         if dag.is_paused_upon_creation is not None:
             orm_dag.is_paused = dag.is_paused_upon_creation
-        orm_dag.bundle_name = bundle_name
         log.info("Creating ORM DAG for %s", dag.dag_id)
         session.add(orm_dag)
         yield orm_dag
@@ -273,8 +270,6 @@ def _update_import_errors(files_parsed: set[str], import_errors: dict[str, str],
 
 
 def update_dag_parsing_results_in_db(
-    bundle_name: str,
-    bundle_version: str | None,
     dags: Collection[MaybeSerializedDAG],
     import_errors: dict[str, str],
     warnings: set[DagWarning],
@@ -312,7 +307,7 @@ def update_dag_parsing_results_in_db(
             )
             log.debug("Calling the DAG.bulk_sync_to_db method")
             try:
-                DAG.bulk_write_to_db(bundle_name, bundle_version, dags, session=session)
+                DAG.bulk_write_to_db(dags, session=session)
                 # Write Serialized DAGs to DB, capturing errors
                 # Write Serialized DAGs to DB, capturing errors
                 for dag in dags:
@@ -351,8 +346,6 @@ class DagModelOperation(NamedTuple):
     """Collect DAG objects and perform database operations for them."""
 
     dags: dict[str, MaybeSerializedDAG]
-    bundle_name: str
-    bundle_version: str | None
 
     def find_orm_dags(self, *, session: Session) -> dict[str, DagModel]:
         """Find existing DagModel objects from DAG objects."""
@@ -372,8 +365,7 @@ def add_dags(self, *, session: Session) -> dict[str, DagModel]:
         orm_dags.update(
             (model.dag_id, model)
             for model in _create_orm_dags(
-                bundle_name=self.bundle_name,
-                dags=(dag for dag_id, dag in self.dags.items() if dag_id not in orm_dags),
+                (dag for dag_id, dag in self.dags.items() if dag_id not in orm_dags),
                 session=session,
             )
         )
@@ -438,8 +430,6 @@ def update_dags(
             dm.timetable_summary = dag.timetable.summary
             dm.timetable_description = dag.timetable.description
             dm.asset_expression = dag.timetable.asset_condition.as_expression()
-            dm.bundle_name = self.bundle_name
-            dm.latest_bundle_version = self.bundle_version
 
             last_automated_run: DagRun | None = run_info.latest_runs.get(dag.dag_id)
             if last_automated_run is None: