tardunge commented on code in PR #63928:
URL: https://github.com/apache/airflow/pull/63928#discussion_r3147723980
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -368,5 +384,129 @@ def view_url(self, name: str, version: str | None = None)
-> str | None:
DeprecationWarning,
stacklevel=2,
)
- bundle = self.get_bundle(name, version)
- return bundle.view_url(version=version)
+ try:
+ bundle = self.get_bundle(name, version)
+ return bundle.view_url(version=version)
+ except ValueError:
+ # Bundle no longer configured, return None
+ self.log.debug("Bundle '%s' is no longer configured, cannot get
view URL", name)
+ return None
+
+ def get_bundle_path_safe(self, name: str) -> Path | None:
+ """
+ Safely get a bundle's path without raising errors if bundle doesn't
exist.
+
+ :param name: The name of the DAG bundle.
+ :return: The bundle path if configured, None otherwise.
+ """
+ try:
+ bundle = self.get_bundle(name)
+ return bundle.path
+ except ValueError:
+ # Bundle no longer configured
+ self.log.debug("Bundle '%s' is no longer configured, cannot get
path", name)
+ return None
+
+ def _parse_config_from_path(self, config_path: str) -> None:
+ """
+ Parse DAG bundle configurations from JSON files in a directory.
+
+ :param config_path: Path to directory containing JSON config files
+ """
+ path = Path(config_path)
+ if not path.exists():
+ self.log.warning("DAG bundle config path does not exist: %s",
config_path)
+ return
+
+ if not path.is_dir():
+ raise AirflowConfigException(f"dag_bundle_config_path must be a
directory, got: {config_path}")
+
+ # Clear old mtime entries before repopulating
+ self._config_path_mtime.clear()
+
+ config_list = []
+ seen_bundle_names: set[str] = set()
+
+ for file_path in path.glob("*.json"):
+ try:
+ config = json.loads(file_path.read_text())
+ if not isinstance(config, dict):
+ self.log.error("Invalid config in %s: Expected dict but
got %s", file_path, type(config))
+ continue
+
+ # Check for duplicate bundle names
+ bundle_name = config.get("name")
+ if bundle_name and bundle_name in seen_bundle_names:
+ self.log.warning(
+ "Duplicate bundle name '%s' found in %s, skipping this
file",
+ bundle_name,
+ file_path,
+ )
+ continue
+ if bundle_name:
+ seen_bundle_names.add(bundle_name)
+
+ config_list.append(config)
+ # Track file modification time for change detection
+ self._config_path_mtime[str(file_path)] =
file_path.stat().st_mtime
+ except json.JSONDecodeError as e:
+ self.log.error("Failed to parse JSON from %s: %s", file_path,
e)
+ except Exception as e:
+ self.log.error("Error reading config file %s: %s", file_path,
e)
+
+ if not config_list:
+ self.log.warning("No valid DAG bundle configs found in %s",
config_path)
+ return
+
+ bundle_config_list = _parse_bundle_config(config_list)
+ if conf.getboolean("core", "LOAD_EXAMPLES"):
+ _add_example_dag_bundle(bundle_config_list)
+
+ # Clear existing config and reload
+ self._bundle_config.clear()
+ for bundle_config in bundle_config_list:
+ if bundle_config.team_name and not conf.getboolean("core",
"multi_team"):
+ raise AirflowConfigException(
+ "DAG bundle configurations from path "
+ "cannot have a team name when multi-team mode is disabled.
"
+ "To enable multi-team, you need to update section `core`
key `multi_team` in your config."
+ )
+
+ class_ = import_string(bundle_config.classpath)
+ self._bundle_config[bundle_config.name] = _InternalBundleConfig(
+ bundle_class=class_,
+ kwargs=bundle_config.kwargs,
+ team_name=bundle_config.team_name,
+ )
+ self.log.info("DAG bundles loaded from path: %s", ",
".join(self._bundle_config.keys()))
+
+ def check_config_path_changes(self) -> bool:
+ """
+ Check if any configuration files have been added, removed, or modified.
+
+ :return: True if changes detected, False otherwise
+ """
+ config_path = conf.get("dag_processor", "dag_bundle_config_path",
fallback="")
+ if not config_path:
+ return False
+
+ path = Path(config_path)
+ if not path.exists() or not path.is_dir():
+ return False
+
+ current_files: dict[str, float] = {}
+ for file_path in path.glob("*.json"):
+ current_files[str(file_path)] = file_path.stat().st_mtime
Review Comment:
Fixed in 368bfa50 — switched _config_path_mtime to dict[str, int] and use
st_mtime_ns in both the parse and the change-detection paths.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
