This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 62641fa9ac0 Fix CLI --bundle-name finding no Dags in Git bundles
(#73377)
62641fa9ac0 is described below
commit 62641fa9ac01cf2443936790fdd1265ebfd57b86
Author: Y-C <[email protected]>
AuthorDate: Fri Oct 16 01:57:51 2026 +0800
Fix CLI --bundle-name finding no Dags in Git bundles (#73377)
Four CLI paths build a DagBag straight from bundle.path without first
calling bundle.initialize(). A LocalDagBundle already has its files on
disk, so the gap is invisible in local development and in the test
suite; a bundle that materialises its files in initialize() - a Git
bundle clones the repo there - is still an empty path at that point.
dags list and dags list-import-errors then report an empty bundle and
no import errors and exit 0, so a CI gate built on them passes while
the Dag files are never read. tasks test, dags test and the --dag-regex
lookups fall through to the "search every configured bundle" path,
which re-fetches and syncs every bundle to the metadata DB and can
return a same-named Dag from a bundle other than the requested one.
The three call sites that already initialize - dags report, dags
reserialize and the fallback loop in get_bagged_dag - show the intended
shape; that fallback loop got its call in #53699, which fixed only the
one occurrence it was reported for.
Co-authored-by: Eason09053360
<[email protected]>
---
.../src/airflow/cli/commands/dag_command.py | 2 +
airflow-core/src/airflow/utils/cli.py | 2 +
.../tests/unit/cli/commands/test_dag_command.py | 69 ++++++++++++++++++++++
airflow-core/tests/unit/utils/test_cli_util.py | 60 +++++++++++++++++++
4 files changed, 133 insertions(+)
diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py
b/airflow-core/src/airflow/cli/commands/dag_command.py
index a108ca5747e..b4dca05b227 100644
--- a/airflow-core/src/airflow/cli/commands/dag_command.py
+++ b/airflow-core/src/airflow/cli/commands/dag_command.py
@@ -577,6 +577,7 @@ def dag_list_dags(args, *, session: Session = NEW_SESSION)
-> None:
for bundle in all_bundles:
if bundle.name in bundles_to_search:
+ bundle.initialize()
bundle_dagbag = BundleDagBag(
bundle.path, bundle_path=bundle.path,
bundle_name=bundle.name
)
@@ -681,6 +682,7 @@ def dag_list_import_errors(args, *, session: Session =
NEW_SESSION) -> None:
for bundle in all_bundles:
if bundle.name in bundles_to_search:
+ bundle.initialize()
bundle_dagbag = BundleDagBag(
bundle.path, bundle_path=bundle.path,
bundle_name=bundle.name
)
diff --git a/airflow-core/src/airflow/utils/cli.py
b/airflow-core/src/airflow/utils/cli.py
index d5af192b350..fb639364364 100644
--- a/airflow-core/src/airflow/utils/cli.py
+++ b/airflow-core/src/airflow/utils/cli.py
@@ -331,6 +331,7 @@ def get_bagged_dag(
manager = DagBundlesManager()
for bundle_name in bundle_names or ():
bundle = manager.get_bundle(bundle_name)
+ bundle.initialize()
with _airflow_parsing_context_manager(dag_id=dag_id):
dagbag = BundleDagBag(
dag_folder=dagfile_path or bundle.path,
@@ -386,6 +387,7 @@ def get_dags(bundle_names: list | None, dag_id: str,
use_regex: bool = False, fr
return [get_bagged_dag(bundle_names=bundle_names, dag_id=dag_id)]
def _find_dag(bundle):
+ bundle.initialize()
dagbag = BundleDagBag(dag_folder=bundle.path, bundle_path=bundle.path,
bundle_name=bundle.name)
matched_dags = [dag for dag in dagbag.dags.values() if
re.search(dag_id, dag.dag_id)]
return matched_dags
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_command.py
b/airflow-core/tests/unit/cli/commands/test_dag_command.py
index ca9c9756fab..11637bafa46 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_command.py
@@ -21,7 +21,10 @@ import argparse
import json
import logging
import os
+import shutil
+from contextlib import contextmanager
from datetime import datetime, timedelta
+from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock
@@ -36,6 +39,7 @@ from airflow import settings
from airflow._shared.timezones import timezone
from airflow.cli import cli_parser
from airflow.cli.commands import dag_command
+from airflow.dag_processing.bundles.base import BaseDagBundle
from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db
from airflow.dag_processing.processor import DagFileParsingResult,
DagFileProcessorProcess
from airflow.exceptions import AirflowException
@@ -81,6 +85,45 @@ jan_6 = DEFAULT_DATE + timedelta(days=5)
dec_27 = DEFAULT_DATE + timedelta(days=-5)
+class _DeferredPathBundle(BaseDagBundle):
+ """Bundle whose files only land on disk in ``initialize()``, like the Git
bundle's clone."""
+
+ def __init__(self, *, source: str, **kwargs) -> None:
+ super().__init__(**kwargs)
+ self._source = Path(source)
+ self._path = self._source.parent / "checkout"
+
+ def initialize(self) -> None:
+ shutil.copytree(self._source, self._path, dirs_exist_ok=True)
+ super().initialize()
+
+ def get_current_version(self) -> None:
+ return None
+
+ def refresh(self) -> None:
+ """Nothing to refresh."""
+
+ @property
+ def path(self) -> Path:
+ return self._path
+
+
+@contextmanager
+def _configure_deferred_path_bundle(dag_file: Path, tmp_path: Path):
+ source = tmp_path / "source"
+ source.mkdir()
+ shutil.copy(dag_file, source)
+ bundle_config = [
+ {
+ "name": "deferred",
+ "classpath": f"{__name__}._DeferredPathBundle",
+ "kwargs": {"source": str(source), "refresh_interval": 0},
+ }
+ ]
+ with conf_vars({("dag_processor", "dag_bundle_config_list"):
json.dumps(bundle_config)}):
+ yield
+
+
class TestCliDags:
parser: argparse.ArgumentParser
@@ -421,6 +464,32 @@ class TestCliDags:
# Rebuild Test DB for other tests
self.setup_class()
+ @conf_vars({("core", "load_examples"): "false"})
+ def test_cli_list_local_dags_initializes_bundle(self, tmp_path,
stdout_capture):
+ args = self.parser.parse_args(
+ ["dags", "list", "--output", "json", "--local", "--bundle-name",
"deferred"]
+ )
+ with _configure_deferred_path_bundle(TEST_DAGS_FOLDER /
"test_example_bash_operator.py", tmp_path):
+ with stdout_capture as temp_stdout:
+ dag_command.dag_list_dags(args)
+ dag_list = json.loads(temp_stdout.getvalue())
+
+ assert [d["dag_id"] for d in dag_list] ==
["test_example_bash_operator"]
+
+ @conf_vars({("core", "load_examples"): "false"})
+ def test_cli_list_import_errors_initializes_bundle(self, tmp_path,
stdout_capture):
+ args = self.parser.parse_args(
+ ["dags", "list-import-errors", "--output", "json", "--local",
"--bundle-name", "deferred"]
+ )
+ with _configure_deferred_path_bundle(TEST_DAGS_FOLDER /
"test_invalid_cron.py", tmp_path):
+ with stdout_capture as temp_stdout:
+ with pytest.raises(SystemExit) as err_ctx:
+ dag_command.dag_list_import_errors(args)
+ errors = json.loads(temp_stdout.getvalue())
+
+ assert err_ctx.value.code == 1
+ assert [e["filepath"] for e in errors] == ["test_invalid_cron.py"]
+
def test_cli_list_dags_custom_cols(self, stdout_capture):
args = self.parser.parse_args(
["dags", "list", "--output", "json", "--columns",
"dag_id,last_parsed_time"]
diff --git a/airflow-core/tests/unit/utils/test_cli_util.py
b/airflow-core/tests/unit/utils/test_cli_util.py
index c95af0626ea..82dce4b668f 100644
--- a/airflow-core/tests/unit/utils/test_cli_util.py
+++ b/airflow-core/tests/unit/utils/test_cli_util.py
@@ -20,6 +20,7 @@ from __future__ import annotations
import ast
import json
import os
+import shutil
import sys
from argparse import Namespace
from contextlib import contextmanager
@@ -56,6 +57,41 @@ from unit.dag_processing.fake_lang_sdk import
fake_coordinator, write_native_fil
pytestmark = pytest.mark.db_test
repo_root = Path(airflow.__file__).parents[1]
+_DEFERRED_BUNDLE_NAME = "deferred-path-bundle"
+_DEFERRED_BUNDLE_DAG_ID = "deferred_path_bundle_dag"
+
+
+class _DeferredPathBundle(BaseDagBundle):
+ """Bundle whose files only land on disk in ``initialize()``, like the Git
bundle's clone."""
+
+ def __init__(self, *, source: Path, **kwargs) -> None:
+ super().__init__(**kwargs)
+ self._source = source
+ self._path = source.parent / "checkout"
+
+ def initialize(self) -> None:
+ shutil.copytree(self._source, self._path, dirs_exist_ok=True)
+ super().initialize()
+
+ def get_current_version(self) -> None:
+ return None
+
+ def refresh(self) -> None:
+ """Nothing to refresh."""
+
+ @property
+ def path(self) -> Path:
+ return self._path
+
+
+def _make_deferred_path_bundle(tmp_path: Path) -> _DeferredPathBundle:
+ source = tmp_path / "source"
+ source.mkdir()
+ (source / "dag.py").write_text(
+ f"from airflow.sdk import DAG\nwith DAG({_DEFERRED_BUNDLE_DAG_ID!r},
schedule=None):\n pass\n"
+ )
+ return _DeferredPathBundle(name=_DEFERRED_BUNDLE_NAME, source=source)
+
class TestCliUtil:
def test_metrics_build(self):
@@ -114,6 +150,30 @@ class TestCliUtil:
with pytest.raises(AirflowException):
cli.get_dags(None, "foobar", True)
+ @pytest.mark.parametrize(
+ "get_matching_dags",
+ [
+ pytest.param(
+ lambda: cli.get_bagged_dag([_DEFERRED_BUNDLE_NAME],
_DEFERRED_BUNDLE_DAG_ID),
+ id="get_bagged_dag",
+ ),
+ pytest.param(
+ lambda: cli.get_dags([_DEFERRED_BUNDLE_NAME],
f"^{_DEFERRED_BUNDLE_DAG_ID}$", use_regex=True)[
+ 0
+ ],
+ id="get_dags_regex",
+ ),
+ ],
+ )
+ @mock.patch("airflow.utils.cli.DagBundlesManager", autospec=True)
+ def test_explicit_bundle_name_is_initialized(self, mock_manager_cls,
tmp_path, get_matching_dags):
+ bundle = _make_deferred_path_bundle(tmp_path)
+ mock_manager_cls.return_value.get_bundle.return_value = bundle
+ mock_manager_cls.return_value.get_all_dag_bundles.return_value = []
+
+ assert get_matching_dags().dag_id == _DEFERRED_BUNDLE_DAG_ID
+ assert bundle.is_initialized
+
@pytest.mark.parametrize(
("given_command", "expected_masked_command", "is_command_list"),
[
