dilnazanlid commented on code in PR #72442:
URL: https://github.com/apache/airflow/pull/72442#discussion_r3951230982
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -661,12 +672,53 @@ def get_all_dag_bundles(self) -> Iterable[BaseDagBundle]:
"""
for name, cfg in self._bundle_config.items():
try:
- yield cfg.bundle_class(name=name, version=None, **cfg.kwargs)
+ bundle = cfg.bundle_class(name=name, version=None,
**cfg.kwargs)
+ bundle._importer_registry = self.get_importer_registry(name)
+ yield bundle
except Exception as e:
self.log.exception("Error creating bundle '%s': %s", name, e)
# Skip this bundle and continue with others
continue
+ def create_importer_registry(
+ self, bundle_name: str, importers_config: list[dict[str, Any]] | None
+ ) -> DagImporterRegistry:
+ """Create and configure a DagImporterRegistry for a bundle with 3-tier
precedence."""
+ from airflow.dag_processing.importers import DagImporterRegistry
+
+ registry = DagImporterRegistry()
+
+ # Global configuration
+ global_importers = conf.getjson("dag_processor",
"dag_importer_configs", fallback=None)
+ if global_importers:
+ if not isinstance(global_importers, list):
+ raise AirflowConfigException(
+ "Section `dag_processor` key `dag_importer_configs` must
be a list "
+ f"but got {global_importers.__class__.__name__}"
+ )
+ self._load_importers_into_registry(registry, global_importers,
context="global configuration")
+
+ # Bundle explicit mapping
+ if importers_config:
+ self._load_importers_into_registry(registry, importers_config,
context=f"bundle '{bundle_name}'")
+
+ return registry
+
+ def _load_importers_into_registry(
+ self, registry: DagImporterRegistry, configs: list[dict[str, Any]],
context: str
+ ) -> None:
+ """Dynamically load and register custom DAG importers."""
+ for importer, extensions in load_dag_importers(configs,
context=context):
+ registry.register(importer, extensions=extensions)
+
+ def get_importer_registry(self, bundle_name: str) -> DagImporterRegistry:
+ """Get the DAG importer registry for a bundle."""
+ if bundle_name not in self._bundle_importers:
+ cfg = self._bundle_config.get(bundle_name)
+ importers_config = cfg.importers if cfg else None
+ self._bundle_importers[bundle_name] =
self.create_importer_registry(bundle_name, importers_config)
+ return self._bundle_importers[bundle_name]
Review Comment:
Correct, it was for the _bundle_config and parsing of the importer classes
in the bundle manager. I added the similar code as in #70805 in here too, to
check it works correctly. Overall, it would be good to submit this(and other 2
PRs) because there is last PR that will need all and use importers from SDK and
move the filesystem path to dag definition.
Do you think your change is going to be submitted soon?
--
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]
