shahar1 commented on code in PR #73209:
URL: https://github.com/apache/airflow/pull/73209#discussion_r4245066482
##########
airflow-core/src/airflow/dag_processing/bundles/provider.py:
##########
@@ -0,0 +1,139 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
1. I'm quite concerned of overloading the term "Provider", which has a very
clear meaning in Airflow, and using it in this context might create some
confusion.
Suggested alternative: "DagBundleSource"? (Like we have AirflowPluginSource)
2. Please add a short "writing a bundle source" (or w/e we choose to call
it) section to dag-bundles.rst
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
Review Comment:
`DagBundlesManager()` now imports bundle classes lazily via
`ConfigDagBundleProvider`, so an invalid classpath raises here instead of
failing the Dag processor at startup, as it did on main.
After `continue`, the bundle remains in `stored`, but cleanup marks it
inactive, removes its team and import errors, and leaves its Dags stale. With
`--bundle-name`, users see "Bundles not found" instead of the actual import
error.
Could we skip only the URL update on failure while keeping the bundle
active? A regression test with an existing bundle and an invalid classpath
would help.
##########
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:
Could you please drop this comment and make sure that the new behavior is
covered with tests?
--
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]
