samraj2k commented on code in PR #73209:
URL: https://github.com/apache/airflow/pull/73209#discussion_r4251260462
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -309,39 +177,76 @@ class DagBundlesManager(LoggingMixin):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
- self._bundle_config: dict[str, _InternalBundleConfig] = {}
- self.parse_config()
+ self._bundle_provider = self._load_bundle_provider()
- def parse_config(self) -> None:
- """
- Get all DAG bundle configurations and store in instance variable.
+ @cached_property
+ def _example_dag_bundle_paths(self) -> dict[str, str]:
+ if not conf.getboolean("core", "LOAD_EXAMPLES"):
+ return {}
+ return _discover_example_dag_bundle_paths()
+
+ @staticmethod
+ def _load_bundle_provider() -> DagBundleProvider:
+ provider_class = conf.getimport("dag_processor", "dag_bundle_provider")
+ if not isinstance(provider_class, type) or not
issubclass(provider_class, DagBundleProvider):
+ raise AirflowConfigException(
+ "The class configured in section `dag_processor` key
`dag_bundle_provider` "
+ "must extend DagBundleProvider."
+ )
+ try:
+ return provider_class()
+ except AirflowConfigException:
+ raise
+ except Exception as e:
+ raise AirflowConfigException(
+ "Could not initialize the Dag bundle provider configured in
section "
+ "`dag_processor` key `dag_bundle_provider`."
+ ) from e
+
+ def _get_provider_bundle_metadata(self) -> tuple[DagBundleMetadata, ...]:
+ """Get validated metadata returned by the configured Dag bundle
provider."""
+ bundle_metadata: dict[str, DagBundleMetadata] = {}
+ for metadata in self._bundle_provider.get_configured_bundle_metadata():
+ if not isinstance(metadata, DagBundleMetadata):
+ raise AirflowConfigException(
+ "Dag bundle providers must return DagBundleMetadata
objects from "
+ "get_configured_bundle_metadata()."
+ )
+ if metadata.name == _example_dag_bundle_name:
+ raise AirflowConfigException(
+ f"Bundle name '{_example_dag_bundle_name}' is a reserved
name. Please choose another name for your bundle."
+ " Example Dags can be enabled with the '[core]
load_examples' config."
+ )
+ if metadata.name in bundle_metadata:
+ raise AirflowConfigException(
+ f"Dag bundle provider returned duplicate bundle name
'{metadata.name}'."
+ )
+ if metadata.team_name and not conf.getboolean("core",
"multi_team"):
+ raise AirflowConfigException(
+ "Dag bundle metadata cannot have a team name when
multi-team mode is disabled. "
+ "To enable multi-team, update section `core` key
`multi_team` in your config."
+ )
+ bundle_metadata[metadata.name] = metadata
- If a bundle class for a given name has already been imported, it will
not be imported again.
+ return tuple(bundle_metadata.values())
- :meta private:
- """
- if self._bundle_config:
- return
+ def get_active_bundle_metadata(self) -> tuple[DagBundleMetadata, ...]:
+ """Get metadata for all active Dag bundles, including example
bundles."""
+ bundle_metadata = {metadata.name: metadata for metadata in
self._get_provider_bundle_metadata()}
+ if not bundle_metadata:
+ return ()
- bundle_config_list = _load_bundle_config_snapshot().configs
- if not bundle_config_list:
- return
+ # Airflow owns example Dag bundles; configured providers do not need
to include them.
Review Comment:
I have dropped the comment, and added a regression test, thanks you!
--
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]
