You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "uranusjr (via GitHub)" <gi...@apache.org> on 2026/09/14 09:09:06 UTC
[PR] Identity-only DAG discovery and byte-importing [airflow]
uranusjr opened a new pull request, #73118:
URL: https://github.com/apache/airflow/pull/73118
Some AIP-85 changes I mentioned I was thinking when reviewing #72369.
Discovery now filters purely by a source's identity rather than its contents. `find_file_dag_definitions` and `list_dag_definitions` match on extension and `.airflowignore`, skip `__pycache__`, and prefer a .py over a side-by-side .pyc, without reading any file or taking a safe_mode flag.
ZipImporter now processes one member at a time instead of a whole archive. It enumerates individual members during discovery, applying the same `__pycache__` and .py/.pyc rules and the ZipSlip defence there, and imports a single member per call, rather than listing whole .zip files and walking their members inside `import_definition`.
Modules are loaded from a definition's bytes. Source and bytecode are read through the definition with custom loaders in place of the file-backed SourceFileLoader and SourcelessFileLoader, so archive members are no longer extracted to a temporary file to be scanned or imported and .py and .pyc members load in place. Accordingly, `might_contain_dag` now takes a definition rather than a path, and PythonDagImporter is parameterized over the file and zip-member definition types it handles.
Tests cover identity-only discovery, import-time skipping for both files and zip members, .py/.pyc dedup, .pyc import for a file and a member, and corrupt-bytecode handling.
cc @dilnazanlid
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4004041274
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -78,41 +135,43 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List Python DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[FileDagDefinition | ZipFileDagDefinition]:
+ """
+ List Python DAG files in a bundle matching supported extensions.
+
+ This does not look for members inside zip archives. Definition discovery
+ of zip archives is done inside :class:`..zip_importer.ZipImporter`
+ instead.
+ """
+ if bundle.path.is_dir():
+ yield from find_file_dag_definitions(bundle.path, self.supported_extensions)
Review Comment:
I am not sure here, but the `list_dag_definitions` method is said to return `FileDagDefinition | ZipFileDagDefinition`, but `find_file_dag_definitions` method in `base.py` only returns `FileDagDefinition` now.
Before, `find_file_dag_definitions` could return all `DagDefinition`s, but now it will return nothing for python files inside the zip (and overall for all zip files as the `find_file_dag_definitions` is used in the `ZipImporter`'s `list_dag_definitions` too).
Should `ZipFileDagDefinition` be added into the `find_file_dag_definitions` as return type? Or maybe make `ZipFileDagDefinition` subclass of `FileDagDefinition` but I am not sure about it, up to you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024218244
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -166,72 +166,73 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipFileDagDefinition]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipFileDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Skipping unreadable ZIP archive %s: %s", archive.path, e)
Review Comment:
I changed `list_dag_definitions` to be able to emit DagImportError objects to the caller as well as successful import results.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4004296722
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -155,68 +209,50 @@ def get_source_code(self, definition: DagDefinition) -> DagSourceCode:
source_code="# Sourceless bytecode (.pyc) — source code not available\n",
language="python",
)
- return DagSourceCode(
- source_code=definition.read_text(encoding="utf-8"),
- language="python",
- )
-
- def might_contain_dag(self, file_path: str | Path, safe_mode: bool = True) -> bool:
- """Check whether a file might contain Airflow DAGs according to safe mode heuristics."""
- if not safe_mode:
- return True
- return might_contain_dag(str(file_path), safe_mode, conf=conf)
+ return DagSourceCode(source_code=definition.read_text(encoding="utf-8"), language="python")
def _load_modules_from_file(
Review Comment:
Good point
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4004041274
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -78,41 +135,43 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List Python DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[FileDagDefinition | ZipFileDagDefinition]:
+ """
+ List Python DAG files in a bundle matching supported extensions.
+
+ This does not look for members inside zip archives. Definition discovery
+ of zip archives is done inside :class:`..zip_importer.ZipImporter`
+ instead.
+ """
+ if bundle.path.is_dir():
+ yield from find_file_dag_definitions(bundle.path, self.supported_extensions)
Review Comment:
I am not sure here, but the method is said to return `ZipFileDagDefinition` too, but `find_file_dag_definitions` only returns FileDagDefinition now. Should `ZipFileDagDefinition` be added into the `find_file_dag_definitions` as return type? As it could return all DagDefinition class children before, now it might return nothing for python files within zip (and overall for all zip files as the `find_file_dag_definitions` is used in the `zip_importer.py` too).
Or maybe make `ZipFileDagDefinition` subclass of `FileDagDefinition` but I am not sure about it, up to you
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -155,68 +209,50 @@ def get_source_code(self, definition: DagDefinition) -> DagSourceCode:
source_code="# Sourceless bytecode (.pyc) — source code not available\n",
language="python",
)
- return DagSourceCode(
- source_code=definition.read_text(encoding="utf-8"),
- language="python",
- )
-
- def might_contain_dag(self, file_path: str | Path, safe_mode: bool = True) -> bool:
- """Check whether a file might contain Airflow DAGs according to safe mode heuristics."""
- if not safe_mode:
- return True
- return might_contain_dag(str(file_path), safe_mode, conf=conf)
+ return DagSourceCode(source_code=definition.read_text(encoding="utf-8"), language="python")
def _load_modules_from_file(
Review Comment:
nit: maybe `_load_modules_from_definition`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024189253
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -166,72 +166,73 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipFileDagDefinition]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipFileDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Skipping unreadable ZIP archive %s: %s", archive.path, e)
+ continue
+
+ member_set = set(member_names)
+ for member_name in member_names:
+ if member_name.endswith("/") or member_name.startswith("__MACOSX/"):
+ continue
+ # ZipSlip defence: reject traversal or absolute member names.
+ member_path = Path(member_name)
+ if member_path.is_absolute() or ".." in member_path.parts:
+ log.warning(
+ "Skipping zip member %r in %s: directory traversal patterns detected",
+ member_name,
+ archive.path,
+ )
+ continue
+
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc,
+ # so a member and its compiled form are never both imported.
+ if "__pycache__" in member_path.parts:
+ continue
+ if member_name.endswith(".pyc") and member_name[:-1] in member_set:
+ continue
+
+ if self._get_internal_importer(member_name) is None:
+ continue
+ yield ZipFileDagDefinition(zip_path=archive.path, file_path=member_name)
Review Comment:
I moved the import context to the DagDefinition subclasses instead to address this.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4015565995
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ return self._definition.read_bytes()
+
+
+class _DefinitionBytecodeLoader(importlib.abc.Loader):
"""
- Importer for Python DAG files.
+ Execute a definition's compiled bytecode (``.pyc``) straight from its bytes.
- This is the default importer registered with the DagImporterRegistry. It handles
- .py files containing Python DAGs.
+ The bytes-based counterpart of :class:`importlib.machinery.SourcelessFileLoader`
+ (which is file-backed); reading through the definition keeps archive members from
+ being extracted just to run them.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_code(self, fullname: str) -> Any:
+ data = self._definition.read_bytes()
+ if len(data) < 16 or data[:4] != importlib.util.MAGIC_NUMBER:
Review Comment:
I just confirmed with the `_classify_pyc` definition in `importlib`.
```python
def _classify_pyc(data, name, exc_details):
"""Perform basic validity checking of a pyc header and return the flags field,
which determines how the pyc should be further validated against the source.
*data* is the contents of the pyc file. (Only the first 16 bytes are
required, though.)
*name* is the name of the module being imported. It is used for logging.
*exc_details* is a dictionary passed to ImportError if it raised for
improved debugging.
ImportError is raised when the magic number is incorrect or when the flags
field is invalid. EOFError is raised when the data is found to be truncated.
"""
magic = data[:4]
if magic != MAGIC_NUMBER:
message = f'bad magic number in {name!r}: {magic!r}'
_bootstrap._verbose_message('{}', message)
raise ImportError(message, **exc_details)
if len(data) < 16:
message = f'reached EOF while reading pyc header of {name!r}'
_bootstrap._verbose_message('{}', message)
raise EOFError(message)
flags = _unpack_uint32(data[4:8])
# Only the first two flags are defined.
if flags & ~0b11:
message = f'invalid flags {flags!r} in {name!r}'
raise ImportError(message, **exc_details)
return flags
```
It seems we could invoke `SourcelessFileLoader.get_code` to ensure the correct logic.
##########
task-sdk/src/airflow/sdk/importers/base.py:
##########
@@ -249,8 +262,14 @@ def find_file_dag_definitions(
if path.suffix.lower() not in supported_exts:
continue
- if safe_mode and not might_contain_dag(str(path), safe_mode, conf=conf):
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc, so a
+ # module and its compiled form are never both imported (a .pyc is used only when
+ # there is no matching .py, i.e. sourceless distribution).
+ if "__pycache__" in path.parts:
continue
+ if path.suffix.lower() == ".pyc" and path.with_suffix(".py").exists():
Review Comment:
Comparing with the merged logic, it seems we can't import the `.pyc` as Kaxil pointed.
https://github.com/apache/airflow/blob/9ecc914fbd00d72617e0b8d107f04d6563d7bbda/task-sdk/src/airflow/sdk/importers/python_importer.py#L59-L85
##########
shared/module_loading/src/airflow_shared/module_loading/dag_file.py:
##########
@@ -37,6 +39,13 @@ def __call__(self, file_path: str, zip_file: zipfile.ZipFile | None = None) -> b
class _ConfLike(Protocol):
def getimport(self, section: str, key: str, **kwargs: Any) -> Any: ...
+ class _DagDefinitionLike(Protocol):
Review Comment:
Would it be better to make this a runtime_checkable protocal? So that we can use safely rely on the `isinstance` as the first check for the `_DagDefinitionLike` then fallback to the str or path like handling.
##########
shared/module_loading/src/airflow_shared/module_loading/dag_file.py:
##########
@@ -102,6 +125,10 @@ def might_contain_dag(
)
if might_contain_dag_callable is None:
- might_contain_dag_callable = might_contain_dag_via_default_heuristic
+ return might_contain_dag_via_default_heuristic(file_path, zip_file=zip_file)
- return might_contain_dag_callable(file_path=file_path, zip_file=zip_file)
+ if isinstance(file_path, (str, os.PathLike)):
+ return might_contain_dag_callable(file_path=file_path, zip_file=zip_file)
+ # Custom callables only accept (file_path, zip_file); let the definition materialize itself.
+ with file_path.as_file() as materialized:
+ return might_contain_dag_callable(file_path=str(materialized), zip_file=None)
Review Comment:
How about having try catch when invoking the `might_contain_dag_callable`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024207851
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ return self._definition.read_bytes()
+
+
+class _DefinitionBytecodeLoader(importlib.abc.Loader):
"""
- Importer for Python DAG files.
+ Execute a definition's compiled bytecode (``.pyc``) straight from its bytes.
- This is the default importer registered with the DagImporterRegistry. It handles
- .py files containing Python DAGs.
+ The bytes-based counterpart of :class:`importlib.machinery.SourcelessFileLoader`
+ (which is file-backed); reading through the definition keeps archive members from
+ being extracted just to run them.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_code(self, fullname: str) -> Any:
+ data = self._definition.read_bytes()
+ if len(data) < 16 or data[:4] != importlib.util.MAGIC_NUMBER:
Review Comment:
Fixed with an additional check. (Also with test case added.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4013531306
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -78,41 +135,43 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List Python DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[FileDagDefinition | ZipFileDagDefinition]:
+ """
+ List Python DAG files in a bundle matching supported extensions.
+
+ This does not look for members inside zip archives. Definition discovery
+ of zip archives is done inside :class:`..zip_importer.ZipImporter`
+ instead.
+ """
+ if bundle.path.is_dir():
+ yield from find_file_dag_definitions(bundle.path, self.supported_extensions)
Review Comment:
I reworked the DagDefinition types to like this:
- One base FileDagDefinition referenced by all dag importers that work on file-like objects.
- FilesystemDagDefinition to represent a “real” file.
- ZipMemberDagDefinition for a member in a zip archive.
PythonDagImporter only interfaces with FileDagDefinition; other importers that will be used as ZipDagImporter’s internal importers will too. Individual importers can provide zip-specific optimization if it wants to, but it’s not required for each importer to know it’s working on zip.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024231259
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ return self._definition.read_bytes()
Review Comment:
Fixed with an extra check.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #73118:
URL: https://github.com/apache/airflow/pull/73118
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024233862
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
Review Comment:
Fixed, the new implementation now matches the old format.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4004994988
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ return self._definition.read_bytes()
+
+
+class _DefinitionBytecodeLoader(importlib.abc.Loader):
"""
- Importer for Python DAG files.
+ Execute a definition's compiled bytecode (``.pyc``) straight from its bytes.
- This is the default importer registered with the DagImporterRegistry. It handles
- .py files containing Python DAGs.
+ The bytes-based counterpart of :class:`importlib.machinery.SourcelessFileLoader`
+ (which is file-backed); reading through the definition keeps archive members from
+ being extracted just to run them.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_code(self, fullname: str) -> Any:
+ data = self._definition.read_bytes()
+ if len(data) < 16 or data[:4] != importlib.util.MAGIC_NUMBER:
Review Comment:
Thinner than `SourcelessFileLoader` in two ways: `_classify_pyc` also validates the PEP 552 flags word, and `_compile_bytecode` rejects a payload that unmarshals to something other than a code object. Without the second, a `.pyc` holding a marshalled string reaches the `exec()` below and runs as source, where the old loader raises `ImportError: Non-code object`. Invalid flags are likewise accepted here and were rejected before. `isinstance(code, types.CodeType)` covers both, and keeps `test_import_corrupt_pyc_captured_as_error` meaningful past the magic number.
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -166,72 +166,73 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipFileDagDefinition]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipFileDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Skipping unreadable ZIP archive %s: %s", archive.path, e)
+ continue
+
+ member_set = set(member_names)
+ for member_name in member_names:
+ if member_name.endswith("/") or member_name.startswith("__MACOSX/"):
+ continue
+ # ZipSlip defence: reject traversal or absolute member names.
+ member_path = Path(member_name)
+ if member_path.is_absolute() or ".." in member_path.parts:
+ log.warning(
+ "Skipping zip member %r in %s: directory traversal patterns detected",
+ member_name,
+ archive.path,
+ )
+ continue
+
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc,
+ # so a member and its compiled form are never both imported.
+ if "__pycache__" in member_path.parts:
+ continue
+ if member_name.endswith(".pyc") and member_name[:-1] in member_set:
+ continue
+
+ if self._get_internal_importer(member_name) is None:
+ continue
+ yield ZipFileDagDefinition(zip_path=archive.path, file_path=member_name)
Review Comment:
`get_file_suffix()` on a `ZipFileDagDefinition` returns the member's suffix, not `.zip`, so `ZipImporter.can_handle(member)` is False and `get_importer(member)` routes it to `PythonDagImporter`, which imports without `_temporary_sys_path(zip_path)`. At b5b0cb1 a cross-member DAG imports fine via `ZipImporter.import_definition` and fails via the registry-resolved importer with `ModuleNotFoundError: No module named 'shared_helper'`. `registry.can_handle(member)` still returns True, so no warning either.
The new docstring promises definitions come back to the emitting importer, but that holds only for the caller, and the one the SDK ships dispatches on suffix. Separate from the return-type thread above: a `FileDagDefinition` subclass would still carry `.py`.
##########
task-sdk/src/airflow/sdk/importers/base.py:
##########
@@ -234,9 +239,17 @@ def get_file_suffix(definition: DagDefinition | str | Path) -> str | None:
def find_file_dag_definitions(
bundle_path: Path,
supported_extensions: Iterable[str],
- safe_mode: bool = True,
-) -> Iterator[DagDefinition]:
- """Find file DAG definitions in a bundle matching given extensions and respecting .airflowignore."""
+) -> Iterator[FileDagDefinition]:
+ """
+ Discover file DAG definitions in a bundle by *identity* alone.
+
+ This walk decides purely from the file's name and path -- extension, ``.airflowignore``,
+ and the Python source/bytecode pairing -- and never reads a file's contents. Deciding
+ whether a discovered file actually contains a DAG is content-based work that belongs to
Review Comment:
Two parse-unit consequences not in the PR body.
No cheap filter is left: `safe_mode` is gone from `list_dag_definitions` and `AbstractDagImporter` has no content pre-check, so `might_contain_dag` is reachable only on `PythonDagImporter`. `find_dag_file_paths` drops what it rejects (`utils/file.py#L114`) and the survivors become the file queue, one `DagFileProcessorProcess` each (`manager.py#L1431`). On 10 example DAGs plus a 41-module `common/` package: 51 definitions here vs 10 from `list_py_file_paths`.
That walk also yields one path per archive, so a zip is one parse unit today; per-member definitions make it 1:N.
Should definitions map 1:1 to parse requests after the rewiring, and where does the queue-size filter live once they do?
##########
task-sdk/src/airflow/sdk/importers/base.py:
##########
@@ -249,8 +262,14 @@ def find_file_dag_definitions(
if path.suffix.lower() not in supported_exts:
continue
- if safe_mode and not might_contain_dag(str(path), safe_mode, conf=conf):
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc, so a
+ # module and its compiled form are never both imported (a .pyc is used only when
+ # there is no matching .py, i.e. sourceless distribution).
+ if "__pycache__" in path.parts:
continue
+ if path.suffix.lower() == ".pyc" and path.with_suffix(".py").exists():
Review Comment:
`.exists()` is a raw stat, so this dedups against the filesystem rather than against what the walk yielded (the zip side compares to `member_set`). With `.airflowignore` matching `only.py` and `only.pyc` shipped beside it, discovery returns nothing, because the ignored `.py` is still on disk. And `PythonDagImporter(extensions=[".pyc"])` loses `sourceless.pyc` to a `sourceless.py` sibling no importer handles. Both go away if the preference is decided over the paths already emitted.
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ return self._definition.read_bytes()
Review Comment:
`get_data` ignores `path` and always returns the definition's own bytes, where `SourceFileLoader.get_data` read the path it was handed. A DAG asking `__loader__.get_data(...)` or `pkgutil.get_data` for a sibling `config.json` gets `b"MARKER = 'i am the dag source'"` here, and the real JSON on the old loader. Raising for a path that isn't the definition's own would at least be loud.
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -166,72 +166,73 @@ def can_handle(self, definition: DagDefinition | str | Path) -> bool:
def list_dag_definitions(
self,
bundle: BaseDagBundle,
- *,
- safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipFileDagDefinition]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipFileDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Skipping unreadable ZIP archive %s: %s", archive.path, e)
Review Comment:
This drops the only user-visible signal for a broken archive. `import_definition` used to raise `DagImportError(error_type="zip_read_error")`, which reaches the import-errors surface; `test_corrupted_zip_file` was relaxed to assert the log warning instead. Discovery already holds the exception, so could it yield something that fails at import, and let the user learn why the archive went missing?
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -56,12 +65,60 @@
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
Review Comment:
Not this line's fault, but it pushes the `:` separator further out: a zip member's `__file__` becomes `archive.zip:member.py`, and `dag.fileloc`/`relative_fileloc` already use that form via `repr()` and `get_relative_loc`. airflow-core only understands the `os.sep` form. `ZIP_REGEX` (`utils/file.py#L40`) yields no archive group for `archive.zip:dag.py`, so `correct_maybe_zipped` and `open_maybe_zipped` pass it through and the later `os.path.isfile` fails.
It came in with #72369, but this change makes per-member definitions the only way a zip DAG gets discovered, so every zip fileloc will be the colon form. Worth settling `/` vs `:` before the rewiring depends on it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4080922419
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -168,70 +172,78 @@ def list_dag_definitions(
bundle: BaseDagBundle,
*,
safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipMemberDagDefinition | DagImportError]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipMemberDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Cannot read ZIP archive %s: %s", archive.path, e)
+ yield DagImportError(
+ source_reference=archive.get_relative_loc(bundle.path),
+ message=f"Failed to read ZIP archive: {e}",
+ error_type="zip_read_error",
+ )
+ continue
+
+ member_set = set(member_names)
+ for member_name in member_names:
+ if member_name.endswith("/") or member_name.startswith("__MACOSX/"):
+ continue
+ # ZipSlip defence: reject traversal or absolute member names.
+ member_path = Path(member_name)
+ if member_path.is_absolute() or ".." in member_path.parts:
+ log.warning(
+ "Skipping zip member %r in %s: directory traversal patterns detected",
+ member_name,
+ archive.path,
+ )
+ continue
+
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc,
+ # so a member and its compiled form are never both imported.
+ if "__pycache__" in member_path.parts:
+ continue
+ if member_name.endswith(".pyc") and member_name[:-1] in member_set:
+ continue
+
+ if (importer := self._get_internal_importer(member_name)) is None:
+ continue
+ member = ZipMemberDagDefinition(zip_path=archive.path, file_path=member_name)
+ if safe_mode and not importer.might_contain_dag(member, safe_mode):
Review Comment:
Addressed in https://github.com/apache/airflow/pull/73118/commits/d2f744b0971e999d15a7c53e5282665accf2f51d
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -42,26 +45,83 @@
DagImportResult,
DagImportWarning,
DagSourceCode,
+ FileDagDefinition,
_normalize_extensions,
find_file_dag_definitions,
get_file_suffix,
)
if TYPE_CHECKING:
from collections.abc import Iterator
- from types import ModuleType
from airflow.dag_processing.bundles.base import BaseDagBundle # noqa: SDK002
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ # The machinery only asks for get_filename(), i.e. the module's own
+ # source. Any other path is a sibling-resource request this bytes-backed
+ # loader can't serve, so fail loud instead of returning the DAG source.
+ if path != self.get_filename(path):
+ raise FileNotFoundError(path)
+ return self._definition.read_bytes()
+
+
+class _DefinitionBytecodeLoader(importlib.abc.Loader):
Review Comment:
Addressed in https://github.com/apache/airflow/pull/73118/commits/d2f744b0971e999d15a7c53e5282665accf2f51d
##########
shared/module_loading/src/airflow_shared/module_loading/dag_file.py:
##########
@@ -102,6 +125,10 @@ def might_contain_dag(
)
if might_contain_dag_callable is None:
Review Comment:
Addressed in https://github.com/apache/airflow/pull/73118/commits/d2f744b0971e999d15a7c53e5282665accf2f51d
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -168,70 +172,78 @@ def list_dag_definitions(
bundle: BaseDagBundle,
*,
safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipMemberDagDefinition | DagImportError]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipMemberDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Cannot read ZIP archive %s: %s", archive.path, e)
+ yield DagImportError(
+ source_reference=archive.get_relative_loc(bundle.path),
+ message=f"Failed to read ZIP archive: {e}",
+ error_type="zip_read_error",
+ )
+ continue
+
+ member_set = set(member_names)
+ for member_name in member_names:
+ if member_name.endswith("/") or member_name.startswith("__MACOSX/"):
+ continue
+ # ZipSlip defence: reject traversal or absolute member names.
+ member_path = Path(member_name)
+ if member_path.is_absolute() or ".." in member_path.parts:
+ log.warning(
+ "Skipping zip member %r in %s: directory traversal patterns detected",
+ member_name,
+ archive.path,
+ )
+ continue
+
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc,
+ # so a member and its compiled form are never both imported.
+ if "__pycache__" in member_path.parts:
+ continue
+ if member_name.endswith(".pyc") and member_name[:-1] in member_set:
Review Comment:
Addressed in https://github.com/apache/airflow/pull/73118/commits/d2f744b0971e999d15a7c53e5282665accf2f51d
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024223166
##########
task-sdk/src/airflow/sdk/importers/base.py:
##########
@@ -249,8 +262,14 @@ def find_file_dag_definitions(
if path.suffix.lower() not in supported_exts:
continue
- if safe_mode and not might_contain_dag(str(path), safe_mode, conf=conf):
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc, so a
+ # module and its compiled form are never both imported (a .pyc is used only when
+ # there is no matching .py, i.e. sourceless distribution).
+ if "__pycache__" in path.parts:
continue
+ if path.suffix.lower() == ".pyc" and path.with_suffix(".py").exists():
Review Comment:
Fixed. `get_code` now checks `isinstance(code, types.CodeType)` after unmarshalling and raises ImportError otherwise.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024223166
##########
task-sdk/src/airflow/sdk/importers/base.py:
##########
@@ -249,8 +262,14 @@ def find_file_dag_definitions(
if path.suffix.lower() not in supported_exts:
continue
- if safe_mode and not might_contain_dag(str(path), safe_mode, conf=conf):
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc, so a
+ # module and its compiled form are never both imported (a .pyc is used only when
+ # there is no matching .py, i.e. sourceless distribution).
+ if "__pycache__" in path.parts:
continue
+ if path.suffix.lower() == ".pyc" and path.with_suffix(".py").exists():
Review Comment:
Fixed by deciding the source-over-bytecode preference over the paths the walk actually emitted.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4047596808
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -168,70 +172,78 @@ def list_dag_definitions(
bundle: BaseDagBundle,
*,
safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipMemberDagDefinition | DagImportError]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipMemberDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Cannot read ZIP archive %s: %s", archive.path, e)
+ yield DagImportError(
+ source_reference=archive.get_relative_loc(bundle.path),
+ message=f"Failed to read ZIP archive: {e}",
+ error_type="zip_read_error",
+ )
+ continue
+
+ member_set = set(member_names)
+ for member_name in member_names:
+ if member_name.endswith("/") or member_name.startswith("__MACOSX/"):
+ continue
+ # ZipSlip defence: reject traversal or absolute member names.
+ member_path = Path(member_name)
+ if member_path.is_absolute() or ".." in member_path.parts:
+ log.warning(
+ "Skipping zip member %r in %s: directory traversal patterns detected",
+ member_name,
+ archive.path,
+ )
+ continue
+
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc,
+ # so a member and its compiled form are never both imported.
+ if "__pycache__" in member_path.parts:
+ continue
+ if member_name.endswith(".pyc") and member_name[:-1] in member_set:
Review Comment:
With `internal_importers={".pyc": PythonDagImporter()}`, an archive containing `dag.py` and `dag.pyc` now yields nothing: the source is unsupported, but its presence suppresses the supported bytecode before importer lookup. Please deduplicate over supported candidates, as filesystem discovery does, and normalize extension case so `dag.PY` and `dag.pyc` do not both survive.
##########
task-sdk/src/airflow/sdk/importers/zip_importer.py:
##########
@@ -168,70 +172,78 @@ def list_dag_definitions(
bundle: BaseDagBundle,
*,
safe_mode: bool = True,
- ) -> Iterator[DagDefinition]:
- """List zip archive DAG definitions in a bundle matching supported extensions."""
- yield from find_file_dag_definitions(bundle.path, self.supported_extensions, safe_mode=safe_mode)
+ ) -> Iterator[ZipMemberDagDefinition | DagImportError]:
+ """
+ List importable members across the bundle's zip archives.
+
+ Each member is yielded as a plain ZipMemberDagDefinition; import_definition
+ re-resolves the internal importer from the member's extension.
+ """
+ for archive in find_file_dag_definitions(bundle.path, self.supported_extensions):
+ try:
+ with zipfile.ZipFile(archive.path) as z:
+ member_names = z.namelist()
+ except Exception as e:
+ log.warning("Cannot read ZIP archive %s: %s", archive.path, e)
+ yield DagImportError(
+ source_reference=archive.get_relative_loc(bundle.path),
+ message=f"Failed to read ZIP archive: {e}",
+ error_type="zip_read_error",
+ )
+ continue
+
+ member_set = set(member_names)
+ for member_name in member_names:
+ if member_name.endswith("/") or member_name.startswith("__MACOSX/"):
+ continue
+ # ZipSlip defence: reject traversal or absolute member names.
+ member_path = Path(member_name)
+ if member_path.is_absolute() or ".." in member_path.parts:
+ log.warning(
+ "Skipping zip member %r in %s: directory traversal patterns detected",
+ member_name,
+ archive.path,
+ )
+ continue
+
+ # Skip compiled-bytecode caches, and prefer source over a side-by-side .pyc,
+ # so a member and its compiled form are never both imported.
+ if "__pycache__" in member_path.parts:
+ continue
+ if member_name.endswith(".pyc") and member_name[:-1] in member_set:
+ continue
+
+ if (importer := self._get_internal_importer(member_name)) is None:
+ continue
+ member = ZipMemberDagDefinition(zip_path=archive.path, file_path=member_name)
+ if safe_mode and not importer.might_contain_dag(member, safe_mode):
Review Comment:
A zip with a valid directory but a bad member CRC raises `BadZipFile` here and terminates the discovery generator, so later valid members are never yielded. I reproduced this against the base, which reports the bad member as an import error and still imports the following Dag. Please catch member-read failures individually, yield a `DagImportError`, and continue.
##########
shared/module_loading/src/airflow_shared/module_loading/dag_file.py:
##########
@@ -102,6 +125,10 @@ def might_contain_dag(
)
if might_contain_dag_callable is None:
Review Comment:
The shipped configuration explicitly defaults to `airflow.utils.file.might_contain_dag_via_default_heuristic`, so `getimport()` returns a callable rather than `None`. Default-config zip discovery therefore still reaches `as_file()` and extracts every scanned member to a temporary file. Please recognize the configured built-in heuristic here and test with the real default configuration.
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -42,26 +45,83 @@
DagImportResult,
DagImportWarning,
DagSourceCode,
+ FileDagDefinition,
_normalize_extensions,
find_file_dag_definitions,
get_file_suffix,
)
if TYPE_CHECKING:
from collections.abc import Iterator
- from types import ModuleType
from airflow.dag_processing.bundles.base import BaseDagBundle # noqa: SDK002
log = logging.getLogger(__name__)
-class PythonDagImporter(AbstractDagImporter):
+class _DefinitionSourceLoader(importlib.abc.SourceLoader):
+ """
+ A SourceLoader that executes a DagDefinition straight from its bytes.
+
+ It needs no file on disk: :meth:`.get_data`` returns the definition's
+ source, and :meth:`get_filename` reports the definition's repr, so
+ ``__file__`` and tracebacks stay meaningful.
+
+ Bytecode caching is left disabled (the inherited ``path_stats`` raises
+ ``OSError``) since it's not particularly useful in dag processors.
+ """
+
+ def __init__(self, definition: DagDefinition) -> None:
+ self._definition = definition
+
+ def get_filename(self, fullname: str) -> str:
+ return repr(self._definition)
+
+ def get_data(self, path: str) -> bytes:
+ # The machinery only asks for get_filename(), i.e. the module's own
+ # source. Any other path is a sibling-resource request this bytes-backed
+ # loader can't serve, so fail loud instead of returning the DAG source.
+ if path != self.get_filename(path):
+ raise FileNotFoundError(path)
+ return self._definition.read_bytes()
+
+
+class _DefinitionBytecodeLoader(importlib.abc.Loader):
Review Comment:
Unlike `SourcelessFileLoader`, this loader has no `is_package()`, so `spec_from_loader()` treats `__init__.pyc` as an ordinary module. A sourceless Dag package containing `from .helper import ID` imports on the base but now fails with `attempted relative import with no known parent package`. Please preserve package detection and cover this case.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024257330
##########
shared/module_loading/src/airflow_shared/module_loading/dag_file.py:
##########
@@ -37,6 +39,13 @@ def __call__(self, file_path: str, zip_file: zipfile.ZipFile | None = None) -> b
class _ConfLike(Protocol):
def getimport(self, section: str, key: str, **kwargs: Any) -> Any: ...
+ class _DagDefinitionLike(Protocol):
Review Comment:
`might_contain_dag` should only be used internally or in controlled paths (admin-approved), so an isinstance check would be too defensive to me. The caller should be responsible for the input type instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024202933
##########
task-sdk/src/airflow/sdk/importers/base.py:
##########
@@ -234,9 +239,17 @@ def get_file_suffix(definition: DagDefinition | str | Path) -> str | None:
def find_file_dag_definitions(
bundle_path: Path,
supported_extensions: Iterable[str],
- safe_mode: bool = True,
-) -> Iterator[DagDefinition]:
- """Find file DAG definitions in a bundle matching given extensions and respecting .airflowignore."""
+) -> Iterator[FileDagDefinition]:
+ """
+ Discover file DAG definitions in a bundle by *identity* alone.
+
+ This walk decides purely from the file's name and path -- extension, ``.airflowignore``,
+ and the Python source/bytecode pairing -- and never reads a file's contents. Deciding
+ whether a discovered file actually contains a DAG is content-based work that belongs to
Review Comment:
I moved the sniffing back to the discovery phase (`list_dag_definitions`) to match the in-use implementation. The sniffing logic would not be as useful in non-Python formats, so we probably need some additional work around it when we reach that part, but at least we should keep the Python file behavior not too different, at least for now.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024179567
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -155,68 +209,50 @@ def get_source_code(self, definition: DagDefinition) -> DagSourceCode:
source_code="# Sourceless bytecode (.pyc) — source code not available\n",
language="python",
)
- return DagSourceCode(
- source_code=definition.read_text(encoding="utf-8"),
- language="python",
- )
-
- def might_contain_dag(self, file_path: str | Path, safe_mode: bool = True) -> bool:
- """Check whether a file might contain Airflow DAGs according to safe mode heuristics."""
- if not safe_mode:
- return True
- return might_contain_dag(str(file_path), safe_mode, conf=conf)
+ return DagSourceCode(source_code=definition.read_text(encoding="utf-8"), language="python")
def _load_modules_from_file(
Review Comment:
I renamed the function to simply `_load_modules` (it’s pretty obvious from tne inputs where it loads from, and the function is internal so we don’t need to be too verbose)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Identity-only DAG discovery and byte-importing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #73118:
URL: https://github.com/apache/airflow/pull/73118#discussion_r4024278485
##########
shared/module_loading/src/airflow_shared/module_loading/dag_file.py:
##########
@@ -102,6 +125,10 @@ def might_contain_dag(
)
if might_contain_dag_callable is None:
- might_contain_dag_callable = might_contain_dag_via_default_heuristic
+ return might_contain_dag_via_default_heuristic(file_path, zip_file=zip_file)
- return might_contain_dag_callable(file_path=file_path, zip_file=zip_file)
+ if isinstance(file_path, (str, os.PathLike)):
+ return might_contain_dag_callable(file_path=file_path, zip_file=zip_file)
+ # Custom callables only accept (file_path, zip_file); let the definition materialize itself.
+ with file_path.as_file() as materialized:
+ return might_contain_dag_callable(file_path=str(materialized), zip_file=None)
Review Comment:
I think the process should crash if the callable is faulty. The callable is supposed to be admin-approved (it’s a part of the installation’s configuration), not arbitrary user code, so an error should be reported very loudly, not caught.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org