AIP-85 DAG importer

Status

Motivation

What problem does it solve? / Why is it needed?

This AIP is a scoped-down, modified part of bigger initiative, which aims to establish extendable DAG parsing controls within Airflow. We’ve strategically divided  it into 2 independent parts to better manage complexity: while this AIP focuses on the implementation of the DAG importer interface, the subsequent AIP covers the overarching architectural framework for parsing controls with addition of interactive DAG processor (AIP-116). This modular approach allows us to implement the core importer mechanism faster and independently of the broader architectural updates.

The primary motivation for this AIP is:

• Many organizations are building customized workflow orchestration solutions on top of Airflow, often preferring different formats(Java, or YAML) to define their DAGs. Currently, these solutions require significant workarounds because Airflow is primarily designed to parse DAGs exclusively from Python files, even when the underlying generation process is a simple, standardized transformation.
• While AIP-72 successfully decoupled task definitions from Python, DAG structural definitions remain strictly bound to the Python language. This limitation is increasingly relevant as industry adoption of non-Python DAG generation tools (such as DAG Factory/Orchestration Pipelines and serverless configurations like AWS MWAA) grows. Allowing these alternative formats to be first-class citizens in Apache Airflow would:
  • Eliminate the need for complex, brittle hacks to support external DAG definitions.
  • Enable native support for diverse DAG generation sources.
  • Align Airflow’s architecture with modern, multi-format workflow requirements.

Benefits

Implementing the DAG importer interface introduces several key operational, scaling, and architectural benefits like:

Elimination of python "bridge" code and CI complexity

• Today, using declarative formats (like YAML via dag-factory or custom JSON schemas) requires maintaining a Python "bridge" file to load configs because Airflow's DAG processor only recognizes .py files. As deployments scale, a single Python bridge file suffers from performance bottlenecks. To scale further, teams are forced to either complicate their CI/CD pipelines to dynamically generate 1-to-1 Python files for every declarative config or devise complex custom sharding schemas. The DAG importer eliminates this entire operational layer, allowing data teams to focus strictly on their configuration files without worrying about underlying DAG processor orchestration quirks.

Airflow-native optimizations for declarative DAGs

• Instead of treating declarative formats as secondary citizens that must pass through an intermediary script, alternative formats become first-class entities. Declarative files (such as individual YAML trees) will be ingested natively, allowing them to potentially benefit from the core DAG processor's internal caching, parallelization, and scheduling optimizations without the performance overhead or translation friction of a homegrown generator framework.

Foundation for multi-language DAG authoring (AIP-108 Alignment)

• AIP-85 directly unblocks multi-language authoring by providing the exact structural interface required to extract DAG definitions from non-Python languages. This creates a clean separation of concerns: an implementation of AIP-85's AbstractDagImporter (e.g., a JavaDagImporter) handles the discovery and parsing of the DAG structure, which then allows components from AIP-108 (like a JavaCoordinator) to natively hook in and manage the actual task execution.

• Note: It is important to clarify that AIP-85 and AIP-108 operate at different levels of the Airflow lifecycle to achieve multi-language support. While AIP-108 focuses on the task execution phase via the Task SDK’s Language Coordinator Layer, this AIP (AIP-85) focuses strictly on the DAG authoring and parsing phase.

Standardized UI visibility ("Code" Tab)

• The improvement cleanly delegates the responsibility of source-text retrieval to the specific importer itself. Whether a user is inspecting a native YAML tree configuration or a non-Python code file managed by an external language SDK, the Airflow Web UI can reliably fetch and display the accurate source definitions without relying on brittle workarounds like overriding dag.fileloc.

Considerations

What change do you propose to make?

The high-level architecture centers on a modular registry-based approach, decoupling DAG discovery from parsing. This enables support for diverse DAG formats beyond Python files, such as YAML or serialized formats, by delegating parsing to specialized, pluggable importer implementations. The architecture for the proposal entails the following flow:

The design of the solution consists of three integral changes: 

• Airflow SDK
• Airflow Configuration
• Airflow Custom Importers

Proposal for Airflow SDK changes

Introduce extensible architecture centered around three primary entities within the Airflow SDK: DagDefinition, AbstractDagImporter and DagImporterRegistry.

DagDefinition 

Instead of operating directly on file paths, importers will interact with a rich DagDefinition object. This abstraction natively supports non-filesystem sources (like ZIP contents or binaries) and separates I/O operations from logical extraction. 

class DagDefinition:
	@property
    def freshness_token(self) -> str: ... 
    
    def read_bytes(self) -> bytes: ... 
    
    def read_text(self, encoding="utf-8") -> str: ...
    
    def as_file(self) -> ContextManager[pathlib.Path]: ...
    
    def __repr__(self) -> str: ...

• freshness_token(): Replaces hard-wired mtime checks with an opaque, generalized token, giving different source types the flexibility to implement their own update-detection logic 
• read_bytes(): Read and return the content of the named resource as bytes.
• read_text(encoding: str): Read and return the content of the named resource as str.
• as_file(): Returns a context manager that yields a pathlib.Path pointing to an existing filesystem object. For file-based definitions, it returns the backing file. For non-filesystem files, it creates a temporary file, yields its path, and cleans it up on exit, preserving backward compatibility for importers that rely heavily on filesystem paths.
• __repr__(): Provides the string representation used by import error and warning objects to properly point the user to the correct location.

AbstractDagImporter

This is the base interface responsible for the discovery, validation, and construction of DAG objects. It serves as an adapter layer that decouples the DAG definition source from the Airflow runtime. Importers are designed to be stateless or trivially constructible, ensuring they do not introduce complexity when managed across IPC boundaries in Airflow’s multi-process environment.

Proposed Interface

class AbstractDagImporter:
	@property
	def supported_extensions(self) -> list[str]: ...

 	def import_file(self, definition: DagDefinition, *, bundle_path: str, bundle_name: str, safe_mode: bool) -> DagImportResult: ... 

 	def can_handle(self, definition: DagDefinition) -> bool: ... 

  	def list_dag_files(self, dir: str, safe_mode: bool) -> Iterator[DagDefinition]: ... 

 	def get_source_code(self, definition: DagDefinition) -> DagSourceCode: ...  

• supported_extensions: A class attribute that defines the file extensions this importer can parse.
• import_file(): A method that parses the target DAG definition and returns a DagImportResult containing parsed DAGs, import errors, and warnings.
• can_handle(): A method that determines if the importer is suitable for a given DAG definition.
• list_dag_files(): A method that traverses the directory to yield compatible DAG definitions for further processing.
• get_source_code(): A method that retrieves the raw source code and its language identifier for the specified DAG definition (this will allow the Web UI to render the native definition in the Code tab).

  • class DagSource:
    	source_code: str
    	language: str

The DagImportResult dataclass will standardize output across different formats:

class DagImportResult:
    definition: DagDefinition
    dags: list[DAG]
    errors: list[DagImportError]
    skipped_definitions: list[DagDefinition]
    warnings: list[DagImportWarning]
	dependencies: list[DagDefinition]

• definition: DagDefinition object representing the source that was parsed. The DAG processor can use definition.__repr__() to log or display its location.
• dags: list of executable DAG objects successfully parsed from the given file path.
• errors: list of fatal parsing issues for the given file path. This should be used strictly for issues that prevent DAG construction entirely (e.g., invalid syntax or missing mandatory fields). The DAG processor will persist these to be surfaced prominently in the Airflow UI.
• warnings: list of non-fatal issues for the given file path, such as the use of deprecated fields. To support frontend UI translation and avoid storing hardcoded user-facing messages directly in the database, custom importers will emit structured DagImportWarning objects.  
  • Note: currently, the warning_type is enum field that supports only specific predefined types. To support custom warning types, the DagImportWarning model has to be adjusted to accept namespaced string field for warning_type field (e.g., "yaml:deprecated_field"), alongside a dictionary of context variables.  
• skipped_definitions: list of DAG definitions that show a clear intent of being a DAG but are ultimately unparsable due to different reasons.
  • Guidance: To prevent flooding the Airflow UI with useless noise, importers must exercise discretion. Completely unrelated definitions should be silently ignored. Definitions appended here will be surfaced in the Airflow UI to give users visibility and prevent hard-to-debug "silent skipping" scenarios.
• dependencies: auxiliary dependencies to monitor for dependency changes and prevent staleness.
  • Note: the DAG Processor will extract the freshness_token  from each of these objects and cache them. During its parsing loop, the DAG Processor will monitor these dependencies; if any token changes, it will automatically trigger a re-parse of the primary definition to prevent stale DAGs. 

DagImporterRegistry

The DagImporterRegistry acts as a centralized service locator and orchestrator and the entire resolution and discovery process is desidned to be fully agnostic to the local filesystem. It maintains a 1:1 mapping between DAG Bundles and their configured importers. When the DagBag attempts to discover DAGs, it queries this registry to identify the correct importer for a given DAG definition.

Proposed Interface

class DagImporterRegistry:
    @property
	def supported_extensions(self) -> list[str]: ...

	def register(self, importer: AbstractDagImporter, extensions: list[str]) -> None: ...

    def get_importer(self, definition: DagDefinition) -> AbstractDagImporter | None: ...

    def can_handle(self, definition: DagDefinition) -> bool: ...

    def supported_extensions(self) -> list[str]: ...

    def list_dag_files(self, bundle: BaseDagBundle, safe_mode: bool) -> Iterator[DagDefinition]: ...

• supported_extensions: A class attribute that defines the extensions for which there is a registered AbstractDagImporter.
• register(): A method that registers the DAG importer to handle the specific extensions.
• get_importer(): A method that gets the registered importer for the DAG definition.
• can_handle(): A method that checks if there is a registered importer that can handle this DAG definition.
• list_dag_files(): A method that lists the DAG definitions in the DAG bundle via registered DAG importers. 

DagBag backwards compatibility

To maintain strict backward compatibility across the Airflow, particularly for thousands of existing unit tests and internal utilities that instantiate DagBag, the DagImporterRegistry will be integrated using an optional dependency injection pattern. The DagBag.__init__ method accepts an optional importer_registry argument. When omitted, the constructor automatically falls back to invoking the global, default-initialized DagImporterRegistry singleton instance, which arrives pre-loaded with the PythonDagImporter, ensuring that legacy tests and standard Python-based DAG environments parse exactly as they did in legacy versions, requiring zero code modifications from end-users or provider developers.

def __new__(cls) -> DagImporterRegistry:
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance._importers = {}
                cls._instance._register_default_importers()
        return cls._instance

def _register_default_importers(self) -> None:
        from airflow.sdk.importers.python_importer import PythonDagImporter

        self.register(PythonDagImporter())

Implemented foundation [as of Jan 2026]

The foundation for the above entities / changes was laid in the following PR.

The foundational elements of this proposal have already been merged into the main branch in the following PR. The following components are currently implemented:

• core: AbstractDagImporter base interface;
• core: DagImporterRegistry for centralized service location;
• core: native PythonDagImporter.

Archive importers (specifically ZipImporter)

Currently, the existing PythonDagImporter owns both the .py and .zip extensions. This design has two critical limitations:

• Python DAG parsing and ZIP archive extraction logic are tightly coupled inside a single importer.
• Users cannot package non-Python DAG formats (like YAML DAGs or custom domain-specific language DAGs) within a .zip archive. The current system is locked into assuming that any .zip file exclusively contains Python modules.

To solve these limitations, we propose introducing a generic composite ZipImporter (and a wider pattern for nested/hierarchical importers). The ZipImporter acts purely as a router: it is responsible only for traversing the .zip archive and delegating the discovery and parsing of the files inside to specialized, configurable "internal" importers (format specified in Airflow Configuration changes below).

Proposed design:

With the introduction of the DagDefinition abstraction, the ZipImporter does not need to rely on default brute-force "extraction-to-temp-dir" flow. Instead, it will utilize a native, lazy-loading interface.

1. When the DAG Processor encounters a .zip  archive, the DagImporterRegistry routes the archive's DagDefinition to the ZipImporter.
2. The ZipImporter reads the ZIP table of contents directly. To protect against resource depletion, it only processes internal files whose suffixes match the keys in the configured internal_importers dictionary.
3. For each valid internal file, the ZipImporter constructs a new, nested DagDefinition object representing that specific ZIP entry. This nested definition implements a __repr__()  that natively formats the archive location (e.g., my_archive.zip:dags/dag_1.java ).
   
4. The ZipImporter selects the correct internal importer (e.g., invoking PythonDagImporter for .py  files and JavaDagImporter for .java  or .jar  files) and passes the nested DagDefinition to the internal importer's import_file()  method.
   1. Lazy extraction: If the internal importer requires access to a physical file on disk (like the PythonDagImporter does), it calls definition.as_filesystem_path()  on the nested definition. The nested DagDefinition handles securely extracting that specific file to a temporary location, yields the pathlib.Path , and cleans it up upon exiting the context manager.
   2. Cross-imports: To support standard Python cross-imports within the ZIP archive (e.g., a dag.py  importing a helpers.py  contained in the same ZIP), the ZipImporter temporarily injects the ZIP archive itself into sys.path.
5. The ZipImporter aggregates the DagImportResult objects from all internal importer runs. Because the nested DagDefinition objects manage their own state and string representations, user-facing error logs, warnings, and source code retrieval inherently remain readable without needing the ZipImporter to manually translate file paths post-parse.
6. As a cleanup step, any sys.path entries injected for the archive are removed.

Proposal for Airflow Configuration changes

This section defines the mapping between DAG bundles and their corresponding importer logic. The configuration can be part of [dag_processor]dag_bundle_config_list, for example:

[
  {
  	"name": "dags-folder",
  	"classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
    "kwargs" {}
    "importers": {
    	"py": {
        	"classpath": "airflow.sdk.importers.python_importer.PythonDagImporter"
        },
        "zip": {
        	"classpath": "airflow.sdk.importers.archive_importer.ZipImporter"
            "kwargs": {
            	"internal_importers": {
                	"py": {
                    	"classpath": "airflow.sdk.importers.python_importer.PythonDagImporter"
                    }
            }
        }
    }
  }
]

Defining this mapping within the bundle configuration allows Airflow to maintain strict isolation between different sources/bundles. This structure ensures that parsing logic is coupled with the bundle's lifecycle, simplifying dependency management and reducing the risk of configuration being applied across diverse environments.

Design considerations:

• Resolution order: Importers are resolved based on file extension matching. If multiple importers support the same extension, the importer will be overridden, and, based on the current implementation, will log a warning. The priority for importers with the same extensions is Bundle Explicit Mapping  > Global Configuration  > Default Importers (e.g., PythonDagImporter).
• Composition: The archive importer pattern (as shown above) demonstrates how importers can be nested. This recursive resolution allows for complex bundle structures (e.g., an archive containing multiple DAG definition types). 

Proposal for Custom Importers

Airflow's configuration will be extended to support custom DAG importers. This allows users to author and register custom DAG parsers natively. For this purpose, the Airflow will dynamically resolve custom importers directly from the Airflow configuration using import_string().

The technical implementation involves two steps: defining the custom importer interface and configuring the class path.

Defining a custom importer interface

Users or developers must create a class that inherits from AbstractDagImporter and implements its core interface. The custom importer dictates how files are read, translated into Airflow DAG objects, and returned via the standardized DagImportResult dataclass. Structure:

from airflow.sdk.importers.base import AbstractDagImporter, DagImportResult

class NewDagImporter(AbstractDagImporter):
	supported_extensions: list[str] = [".extension1", ".extension2"]

    def can_handle(self, definition: DagDefinition) -> bool:
        ...

    def import_file(
        self, definition: DagDefinition, *, bundle_path: str, bundle_name: str, safe_mode: bool
    ) -> DagImportResult:
        ...

Configuring the Importer via Airflow Configuration

Custom importers are registered directly in the Airflow configuration file (or via environment variables). Airflow will dynamically instantiate the configured class path at runtime.

[dag_processor]
dag_importer_configs = [
    "custom.importers.NewDagImporter",
    "another_folder.importers.CustomYamlImporter"
]

When the DAG processor initializes, it loops through the specified class paths, imports them using import_string, and registers them to handle incoming bundle files based on their can_handle logic.

Considerations

Performance

To validate that the architecture introduces no latency regressions, and to quantify the expected performance gains of native declarative parsing, the following dimensions and scenarios will be measured as part of the performance tests:

• Number of DAG files
• Number of DAGs in single file
• Number of accesses to other parse-time parameters

Given that the initialization of the DagBag with DagImporterRegistry and loading of importer classes, some small performance regression is anticipated. At the same time alternative parsers will provide the performance improvement in themselves (e.g. by avoiding unnecessary parsing) which can overweight any anticipated regression.

Comparative benchmarking: direct performance comparison measuring the time and resource overhead to parse N declarative DAGs (e.g., YAML) natively via a custom importer, versus parsing the same N DAGs using legacy Python generator patterns (such as DAG Factory).

What defines this AIP as "done"?

This AIP will be considered “done” when the PR with necessary well-documented solution is created and the language-specific custom DAG importers can be registered. 

Overall, the deliverables for this AIP include:

• AbstractDagImporter interface: A new pluggable, public interface allowing custom plugins to transform alternative bundle content into executable Airflow DAG objects.

  • base version implemented; may be changed based on functional needs

• PythonDagImporter: A native implementation migrating legacy DagBag processing logic into this new architecture, ensuring standard Python files continue to parse cleanly.

  • base version implemented; may be changed based on functional needs

• Bundle & Processor integration: Extension of the Airflow 3 Bundle interface to dynamically map and instantiate designated importers per bundle, and full integration within the DAG processor.

• UI Source retrieval & Syntax highlighting: Core API support for methods like get_source_code() and language hinting. The UI will be updated to display the native format and apply the correct syntax highlighting in the 'Code' tab. 

• UI visibility for parsing metadata: The Airflow UI will be updated to consume and display the extended metadata provided by DagImportResult . This implementation includes adding UI components (such as banners or dedicated views) to surface:
  • Skipped files: Giving users visibility into files rejected by the importer's safe_mode  heuristic.
  • Warnings: The existing dag_warning table mechanism will be extended to support custom importer warnings. This requires relaxing the DagWarningType column from a strict database Enum to a standard string column to support namespaced warning types (e.g., "yaml:unrecognized_key"). Additionally, the table will need to store the warning context (e.g., {"field_name": "field_value"})to enable dynamic i18n translation of these messages in the frontend.
  • Errors: Continuing to display critical parsing failures accurately.

Out of Scope 

• DAG reconciliation
  The logic determining how and when a DAG record is purged from the system (for example, if its underlying definition is deleted from the bundle) is explicitly out of scope for the importer contract. Determining whether a DAG record should be removed belongs entirely to the DAG bundle lifecycle management and the DAG Processor. Importers remain strictly focused on discovery and translation.

Known limitations

• Code tab permissions for multi-DAG files: It is a common pattern to define multiple DAGs in a single file. Currently, the Airflow Code tab only renders the file if the user has CODE level permissions for every DAG defined within it. Solving this RBAC granularity issue is deferred to future work, as this limitation is inherited from how Airflow currently handles multi-DAG Python files.