tardunge commented on code in PR #63928:
URL: https://github.com/apache/airflow/pull/63928#discussion_r3147690333
##########
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
Review Comment:
duplicate comment.
--
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]
