dilnazanlid opened a new pull request, #72442:
URL: https://github.com/apache/airflow/pull/72442
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
**Project Board:** [AIP-85 Dag Importer (Project
#672)](https://github.com/orgs/apache/projects/672)
This pull request is one of the incremental parts of the **[AIP-85: DAG
Importer](https://cwiki.apache.org/confluence/x/_Q7OEg)** 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?
- [X] Yes
Generated-by: Google Gemini following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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]
