dilnazanlid opened a new pull request, #72369:
URL: https://github.com/apache/airflow/pull/72369
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
### Description
* **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.
It restructures the DAG importer base abstractions and moves them into the
**Task SDK** (`task-sdk/src/airflow/sdk/importers/`), while migrating shared
file-parsing utilities into `airflow_shared.module_loading`.
### Motivation & Background
Under the Airflow 3 architecture, the Task SDK (`airflow.sdk`) serves as the
lightweight DAG authoring and execution boundary, while DAG file processing and
bundling operate without workers requiring direct database access.
To support pluggable DAG importers across various formats without creating
tight coupling, the fundamental DAG importer interfaces, definitions,
registries, and built-in importers (`PythonDagImporter`, `ZipImporter`) must
live inside the Task SDK.
### Summary of Changes
#### 1. Shared Module Loading (`shared/module_loading`)
- **Extracted Dag file heuristics**: Moved `get_unique_dag_module_name`,
`might_contain_dag`, and `might_contain_dag_via_default_heuristic` from
`airflow.utils.file` into `airflow_shared.module_loading.dag_file`.
- **Backward Compatibility**: Maintained backward compatibility in
`airflow.utils.file` via module attribute fallbacks/deprecation paths.
- **Added Tests**: Added unit tests in
`shared/module_loading/tests/module_loading/test_dag_file.py`.
#### 2. Task SDK Importer Abstractions
(`task-sdk/src/airflow/sdk/importers/`)
- **`DagDefinition` & `FileDagDefinition`**:
- Abstract base class representing a Dag source, with an opaque
`freshness_token` (used for cache invalidation and change detection), relative
location resolution, direct content reading (`read_bytes`, `read_text`), and a
context manager for file access (`as_file()`).
- `FileDagDefinition`: Concrete implementation backed by local filesystem
files.
- **Diagnostics & Results**:
- `DagImportError`: Structured error reporting capturing source
references, line/column numbers, error types, contextual snippets, suggestions,
and stack traces.
- `DagImportWarning`: Non-fatal warnings captured during Dag importing.
- `DagImportResult`: Encapsulates imported Dags, errors, skipped
definitions, warnings, and dependencies.
- `DagSourceCode`: Data structure containing raw source code and language
identifier for Code view presentation.
- **`AbstractDagImporter`**: Base interface specifying
`supported_extensions`, `import_definition()`, `can_handle()`,
`list_dag_definitions()`, and `get_source_code()`.
- **`DagImporterRegistry`**: Thread-safe singleton registry mapping file
extensions to their corresponding importer implementations, with default
registration for `.py` and `.zip`, conflict warnings, and test isolation reset
support.
- **`PythonDagImporter`**:
- Importer for Python files (`.py`) supporting `safe_mode` heuristics,
import timeouts, module execution with unique prefix isolation
(`unusual_prefix_`), and top-level `DAG` / `DagContext` extraction with bundle
metadata tracking.
- Implements `get_source_code` returning Python source.
- **`ZipImporter` & `ZipFileDagDefinition`**:
- Composite importer handling `.zip` archives.
- Protects against **ZipSlip** directory traversal attacks.
- Extracts archive members into isolated temporary storage, safely manages
`sys.path`, and routes archive members to registered internal importers (such
as `PythonDagImporter`).
- Implements archive-aware `freshness_token` and source code extraction.
#### 3. Test Coverage
- `task-sdk/tests/task_sdk/importers/test_python_importer.py`: Tests for
importing valid Dags, syntax error reporting, skipping non-Dag files, bundle
metadata propagation, and source code retrieval.
- `task-sdk/tests/task_sdk/importers/test_registry.py`: Tests for singleton
lifecycle, default registrations, extension resolution, case insensitivity, and
registry resets.
- `task-sdk/tests/task_sdk/importers/test_zip_importer.py`: Tests for ZIP
Dag importing, ZipSlip directory traversal prevention, source code extraction
(whole archive and archive member), and composite freshness tokens.
- `task-sdk/tests/task_sdk/docs/test_public_api.py`: Updated public API
verification to include `airflow.sdk.importers`.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes
Generated-by: Antigravity 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]
