[AIP-85] Move DAG importer abstractions and registry into Task SDK

[dilnazanlid]

Description

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

This pull request is one of the incremental parts of the AIP-85: DAG Importer 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?

• Yes

Generated-by: Antigravity following the guidelines

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack