[AIP-85] Add configuration support for custom DAG importers

[dilnazanlid]

Project Board: AIP-85 Dag Importer (Project #672)

This pull request is one of the incremental parts of the AIP-85: DAG Importer implementation.

Description

This pull request is part of the implementation of AIP-85: DAG Importer. It introduces configuration-driven registration and resolution for custom Dag importers at both the global and Dag bundle levels.

Prior to this change, Dag file discovery and parsing were tied to built-in Python file discovery (list_py_file_paths), and the DagImporterRegistry operated strictly as a singleton with hardcoded defaults. This PR introduces a configurable, tiered Dag importer resolution mechanism:

1. Shared Module Loading (airflow_shared.module_loading.dag_importers):

   • Added load_dag_importers() utility to dynamically instantiate importer classes via classpath and kwargs.
   • Supports explicit extensions configuration per importer, falling back to the importer's supported_extensions attribute or method.
   • Validates configuration formats and raises AirflowConfigException on malformed inputs.

2. 3-Tier Importer Registry Precedence in Dag Bundles (DagBundlesManager):

   • Extended Dag bundle schemas (_ExternalBundleConfig, _InternalBundleConfig) to support an optional importers list.
   • Established precedence hierarchy when building each bundle's DagImporterRegistry:
     1. Bundle configuration (bundle.importers): Importers explicitly declared for that specific bundle (highest priority).
     2. Global configuration ([dag_processor] dag_importer_configs): System-wide custom importers configured via Airflow config.
     3. Default built-ins: Standard PythonDagImporter handling .py and .zip files (baseline fallback).
   • Added bundle.importer_registry property to BaseDagBundle and cached per-bundle registries in DagBundlesManager.

3. Instantiable DagImporterRegistry:

   • Refactored DagImporterRegistry from a rigid process-wide singleton into an instantiable class, allowing isolated registries per Dag bundle while preserving get_importer_registry() for global access.
   • Enhanced register() to allow custom extension lists and log informative warnings when an extension mapping is overridden by a higher-priority importer.

4. Integration with File Discovery and DagBag:

   • Updated DagFileProcessorManager._find_files_in_bundle() to discover Dag files dynamically using bundle.importer_registry.list_dag_files().
   • Updated DagBag to accept and utilize bundle-scoped importer_registry instances during Dag parsing and collection.

5. Tests:

   • Added unit tests for dynamic importer loading and validation in airflow_shared.module_loading.
   • Added tests in test_dag_bundle_manager.py verifying the 3-tier precedence rules, extension overrides, nested composite importers, and error handling.
   • Updated test_registry.py and test_manager.py to cover independent registry instances and bundle discovery.

---

Was generative AI tooling used to co-author this PR?

• Yes

Generated-by: Google Gemini following the guidelines

[SameerMesiah97]

Not an expert here but this looks fairly clean from my perspective but we need to wait for CI to be triggered as there is a high risk of breaking changes here. I have left some comments.

[SameerMesiah97]

Correct me if I am wrong (I am not as familiar with the DAG processor), but it looks like explicitly configured extensions are only used for registry lookup, not file discovery. For example, configring PythonDegImporter for "custom" makes get_importer("dag.custom") succeed, but list_dag_files() delegates to PythonDagImporter.list_dag_files(), which only discovers Python files and ZIP archives. Would custom files therefore never reach parsing?

[dilnazanlid]

Good catch!

Currently, explicit extensions in the configuration only register entries in DagImporterRegistry._importers (affecting get_importer() and can_handle()), but registry.list_dag_files() delegates directly to importer.list_dag_files().

Because AbstractDagImporter.list_dag_files() inspects self.supported_extensions() (which is a @classmethod returning class-level defaults), files with custom configured extensions like dag.custom are filtered out during directory discovery. Consequently, they are never enqueued by DagFileProcessorManager and never reach parsing during normal bundle execution.

Proposed fix:

• DagImporterRegistry.list_dag_files() performs a single walk of the bundle directory.
• Each file is checked against registry.get_importer(file_path). If matched, content validation / safe mode is delegated to importer.might_contain_dag(file_path, safe_mode).
• We associate the configured extensions(global or bundle-level) with the importer instance (importer.set_configured_extensions()) so that importer.can_handle() and importer.supported_extensions stay consistent with the registry.

This will fix file discovery for custom extensions, avoids running N redundant filesystem walks across multiple importers, and strictly enforces extension override precedence.

WDYT? I will implement it for this change into the dag_processing/importers base classes. However, the importers are moved into the SDK in #72369 along with DagDefinition abstraction replacing the filesystem path approach, so it will also be reflected there.

[dilnazanlid]

Added the changes specified above, seems like it solves the problem and tried to make them follow the best practice, PTAL

[SameerMesiah97]

@jason810496

Could you trigger CI?