You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dilnazanlid (via GitHub)" <gi...@apache.org> on 2026/10/01 11:37:25 UTC
[PR] Discover Dag definitions through the Task SDK importers [airflow]
dilnazanlid opened a new pull request, #74020:
URL: https://github.com/apache/airflow/pull/74020
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
* **Project Board:** [AIP-85 Dag Importer (Project #672)](https://github.com/orgs/apache/projects/672)
* 4th PR for rewiring the Dag processing process with `sdk/importers`:
------------
TheDag processor manager still found files with `list_py_file_paths`, which only returns `.py` and `.zip` files. Files handled by a custom importer configured in `dag_importer_configs` were never queued for parsing. Discovery now uses the importer registry, so anything a configured importer can handle is picked up.
Some importers return definitions nested inside a file, such as members of an archive. The manager still queues and parses the enclosing file once. The member locations are kept on `DagFileInfo.definition_locs`, and `deactivate_deleted_dags` and `clear_orphaned_import_errors` match against them. Before, a separate zip-only pass reopened every archive to collect member paths. The `deactivate_deleted_dags(bundle_name, present)` signature is unchanged.
Other fixes for nested definitions:
- If listing a bundle fails, the error is logged and the bundle is skipped. It used to be treated as an empty bundle, which deactivated all of its Dags.
- Stale-Dag detection now matches a Dag nested at any depth in a file that is still present. Before, it only checked one parent level.
- `DagBag.get_dag` / `process_file` now reprocess a nested definition from its own location. Before, the path had to be mapped back to a `.zip` (`correct_maybe_zipped`).
- `list_py_file_paths` and `find_dag_file_paths` are removed from `airflow.utils.file`. They have no remaining callers in this repo.
------------
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164850685
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +990,47 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
+ if (path := find_enclosing_file(bundle.path / rel_fileloc)) is None:
+ self.log.warning(
+ "Ignoring %r listed in bundle %s: no file in the bundle holds it", item, bundle.name
+ )
+ continue
+ definition_locs[path.relative_to(bundle.path)].add(rel_fileloc)
Review Comment:
This could trigger a ValueError is `path` is not inside `bundle.path`. I’d do something like
```python
try:
rel_path = Path(os.path.normpath(path)).relative_to(bundle.path)
except ValueError:
self.log.warning(
"Ignoring %r listed in bundle %s: its file resolves outside the bundle", item, bundle.name
)
continue
```
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr merged PR #74020:
URL: https://github.com/apache/airflow/pull/74020
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164850685
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +990,47 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
+ if (path := find_enclosing_file(bundle.path / rel_fileloc)) is None:
+ self.log.warning(
+ "Ignoring %r listed in bundle %s: no file in the bundle holds it", item, bundle.name
+ )
+ continue
+ definition_locs[path.relative_to(bundle.path)].add(rel_fileloc)
Review Comment:
This could trigger a ValueError is `path` is not inside `bundle.path`. I’d do something like
```python
try:
rel_path = Path(os.path.normpath(path)).relative_to(bundle.path)
except ValueError:
self.log.warning(
"Ignoring %r listed in bundle %s: its file resolves outside the bundle",
item,
bundle.name,
)
continue
```
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164452987
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
Review Comment:
I went with the docstring change - the registry can't apply the right check for composites anyway, since it pairs a zip member with ZipImporter while the meaningful check is the member importer's. Thanks!
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
+ if (path := find_enclosing_file(bundle.path / rel_fileloc)) is not None:
Review Comment:
Fixed, added warning log. (although this should become obsolete with the #73457 where the files are fully replaced by dag definition objects)
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4157624391
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
Review Comment:
On main, `find_dag_file_paths` wrapped each file's `might_contain_dag` in a `try/except` and skipped the bad file. `PythonDagImporter.list_dag_definitions` calls `might_contain_dag` (and through it `read_bytes()`) with no guard, so a `PermissionError`, a file deleted mid-walk, or a `might_contain_dag_callable` that raises on one file ends the generator. The `except` at line 962 can't resume it, so the whole bundle is skipped on every refresh. I reproduced it with a bundle holding `good.py` plus a `chmod 000` file: main lists `['good.py']`, this branch raises `PermissionError`.
`ZipImporter` already isolates unreadable members by yielding a `DagImportError`. Doing the same per file in `PythonDagImporter.list_dag_definitions` would queue the file, surface an import error for it, and keep the rest of the bundle. (#74029 keeps the per-file guard, for what it's worth.)
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -950,10 +957,12 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._bundle_versions[bundle.name] = version_after_refresh
self._bundle_version_data[bundle.name] = version_data_after_refresh
- found_files = {
- DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path)
- for p in self._find_files_in_bundle(bundle)
- }
+ try:
+ found_files = self._find_files_in_bundle(bundle)
Review Comment:
`update_bundle_state(..., version=version_after_refresh)` and `self._bundle_versions[bundle.name] = version_after_refresh` (line 957) both run before this `try`, so by the time the listing raises the processor has already recorded the new version. On the next refresh `previously_seen` is true and the version hasn't moved, so the "version not changed" short-circuit above hits `continue` before we list again. For a git bundle that means the new files in that commit are never queued until another commit lands, and if it happens on the first refresh after startup nothing in the bundle gets parsed by this processor. A priority parse request doesn't get it out either, since `_force_refresh_bundles` only bypasses `should_skip_refresh`. On main the exception crashed the loop and the restart re-listed the bundle.
Could the `except` pop `self._bundle_versions[bundle.name]` and `self._bundle_version_data[bundle.name]`, or could the version only be advanced once the listing succeeds? The comment at line 951 already says `_bundle_versions` is only advanced on success.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
Review Comment:
The importer is discarded here and every definition it lists is kept, so an importer's `might_contain_dag` override only takes effect if its own `list_dag_definitions` calls it. Only `PythonDagImporter` and `ZipImporter` do. The `AbstractDagImporter.might_contain_dag` docstring tells importer authors that overriding it means "obvious non-DAG sources are dropped during discovery".
I checked with a minimal `.yaml` importer that lists through `find_file_dag_definitions` and overrides only `might_contain_dag`: in safe mode a `not_a_dag.yaml` containing `foo: bar` is queued next to `pipeline.yaml`, so every non-Dag YAML file in a bundle gets its own parse process. #74029 gates each file through its importer's `might_contain_dag`, so the two PRs currently disagree on who owns this check. Should the gate live here (`for importer, item in ...`, skipping non-error items where `not importer.might_contain_dag(item, safe_mode)`), or should the base docstring say `list_dag_definitions` has to apply it itself? Gating here would sniff `.py` files twice, so the docstring route may be cleaner.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
Review Comment:
This is now the first place the manager builds a bundle's registry, and it runs inside the loop, after `before_run()` has called `gc.freeze()`. `list_dag_definitions` also materialises any configured importer specs, so a custom importer's module and its dependencies get imported here too. On Linux the parse child is a bare fork and `DagBag.__init__` picks up the same cached object from `get_importer_registry`, so the children inherit these objects outside the frozen set. Would it be worth warming the registry for each bundle (pending specs included) in `before_run` ahead of the freeze, so the parent's copy is the shared one?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
+ if (path := find_enclosing_file(bundle.path / rel_fileloc)) is not None:
Review Comment:
An item with no enclosing file is dropped here without a log. That covers a custom importer yielding a definition that isn't backed by a file (the `DagDefinition` ABC allows that) and a `DagImportError` whose `source_reference` isn't a bundle path, in which case the discovery error is lost as well. A `log.warning` in an `else` branch would tell the importer author why their Dags never show up.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -3887,6 +3964,32 @@ def test_refresh_dag_bundles_update_bundle_state_failure_still_scans_files(self)
# iteration will see a version mismatch and re-refresh rather than skip incorrectly
assert "mock_bundle" not in manager._bundle_versions
+ def test_refresh_dag_bundles_discovery_failure_keeps_known_files_and_dags(self):
Review Comment:
`_make_refresh_bundle()` defaults to `supports_versioning=False`, so this test never reaches the stuck-version path from the comment on line 961. Parametrizing over `supports_versioning` and asserting that a second `_refresh_dag_bundles` call lists again would pin that fix.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
Review Comment:
#74029 changes this same discovery path in `manager.py` and `utils/file.py`, with a different approach. Could the description link it, so the two can be reconciled before either merges? The comments on lines 991 and 996 cover where they currently differ (the per-file error guard, and who calls an importer's `might_contain_dag`).
##########
airflow-core/src/airflow/dag_processing/dagbag.py:
##########
@@ -276,9 +276,7 @@ def get_dag(self, dag_id, *, session: Session = NEW_SESSION):
self.dags.pop(dag_id, None)
if dag is None or is_expired:
# Reprocess source file.
- found_dags = self.process_file(
- filepath=correct_maybe_zipped(orm_dag.fileloc), only_if_updated=False
- )
+ found_dags = self.process_file(filepath=orm_dag.fileloc, only_if_updated=False)
Review Comment:
Nothing pins this change. With `correct_maybe_zipped` restored here the related `get_dag` / refresh / zip tests all still pass, because `test_refresh_packaged_dag` only counts imports of `test_zip.py`, so re-importing the whole archive and re-importing one member look the same. Asserting that the other archive members are not re-imported would bind it.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
Review Comment:
Two user-visible changes come in through the registry here that the description doesn't mention:
- Sourceless `.pyc` files are now queued for parsing. `PythonDagImporter.supported_extensions` is `[".py", ".pyc"]` and `find_file_dag_definitions` keeps a `.pyc` with no `.py` sibling, while the old scanner only took `.py` files and real zips.
- A corrupt `.zip` is now queued with a `zip_read_error` import error from `ZipImporter.list_dag_definitions`. Before, `zipfile.is_zipfile` returned false and the file was skipped without a trace.
Both look like reasonable behaviour, but users will see new import errors after upgrading, so they are worth a line in the description.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -950,10 +957,12 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._bundle_versions[bundle.name] = version_after_refresh
self._bundle_version_data[bundle.name] = version_data_after_refresh
- found_files = {
- DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path)
- for p in self._find_files_in_bundle(bundle)
- }
+ try:
+ found_files = self._find_files_in_bundle(bundle)
+ except Exception:
+ # Treating a failed listing as an empty bundle would deactivate all of its Dags.
Review Comment:
This comment and the PR description both say a failed listing used to be treated as an empty bundle, but that isn't what main does. There `_find_files_in_bundle` ran without a `try`, and `_run_parsing_loop` calls `_refresh_dag_bundles` without one too, so an exception from the listing left the parsing loop. The only empty-bundle case on main was a missing bundle directory, where `list_py_file_paths` returns `[]`. Could the comment and the description describe what the `except` does now, rather than an old behaviour it replaces?
##########
airflow-core/tests/unit/utils/test_file.py:
##########
@@ -181,38 +180,6 @@ def test_get_modules_from_invalid_file(self):
assert len(modules) == 0
- def test_list_py_file_paths(self, test_zip_path):
Review Comment:
This was the only test checking that discovery over a real folder honours `.airflowignore`. With `find_path_from_directory` in `airflow.sdk.importers.base` swapped for a plain `os.walk`, the core and task-sdk importer tests all still pass. A `_find_files_in_bundle` test with an `.airflowignore` in `tmp_path` (one ignored `.py`, one ignored `.zip`, one kept file) would cover it. Small thing in the same file: `TEST_DAG_FOLDER` (line 38) is now unused, and `TestListPyFilesPath` is still named after the removed function.
##########
airflow-core/src/airflow/utils/file.py:
##########
@@ -32,7 +32,6 @@
might_contain_dag as might_contain_dag,
might_contain_dag_via_default_heuristic as might_contain_dag_via_default_heuristic,
)
-from airflow.configuration import conf
log = logging.getLogger(__name__)
Review Comment:
`import logging` and `log` have no users now that `find_dag_file_paths` is gone.
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164454542
##########
airflow-core/src/airflow/utils/file.py:
##########
@@ -32,7 +32,6 @@
might_contain_dag as might_contain_dag,
might_contain_dag_via_default_heuristic as might_contain_dag_via_default_heuristic,
)
-from airflow.configuration import conf
log = logging.getLogger(__name__)
Review Comment:
deleted
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164945918
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +990,47 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
+ if isinstance(item, DagImportError):
+ # Importers report a source either absolutely or relative to the bundle.
+ rel_fileloc = os.path.relpath(bundle.path / item.source_reference, bundle.path)
+ else:
+ rel_fileloc = item.get_relative_loc(bundle.path)
+ if (path := find_enclosing_file(bundle.path / rel_fileloc)) is None:
+ self.log.warning(
+ "Ignoring %r listed in bundle %s: no file in the bundle holds it", item, bundle.name
+ )
+ continue
+ definition_locs[path.relative_to(bundle.path)].add(rel_fileloc)
Review Comment:
Thanks, I moved it outside even before `find_enclosing_file` because it could cause an issue.
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164450813
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
Review Comment:
There is the ADR to move the whole process model to the importers in #73457 but I included `warm_importers()` method into the registry to have it here. As the importers are experimental in 3.4. and there are changes planned in the ADR - I think it should be okay.
cc @uranusjr @jason810496
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
Review Comment:
Thanks, I adde them in the PR description. Also, found another user visible change - zip archives without explicit `.zip` extension (e.g. `.egg`) are no longer discovered so they may be ignored unless explicitly added into the importer config `extensions` field for ZipImporter.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
+ """
+ List the files to parse in a bundle through its importers.
+
+ A file holding several Dag definitions (a zip archive, for instance) is parsed as one.
+ """
+ self.log.info("Searching for Dag definitions in %s at %s", bundle.name, bundle.path)
+ registry = get_importer_registry(bundle.name)
+ definition_locs: defaultdict[Path, set[str]] = defaultdict(set)
+ for _, item in registry.list_dag_definitions(bundle, safe_mode=self.dag_discovery_safe_mode):
Review Comment:
Thanks, now python importer `list_dag_definitions` guards each file the way ZipImporter does - unreadable file is yielded as a `DagImportError`, so it's still queued, the error surfaces, while the rest of the bundle is listed. I feel like it is suitable as the whole `might_contain_dag` method is defined on the importer level - it should be responsible for using.
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164630875
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -3887,6 +3964,32 @@ def test_refresh_dag_bundles_update_bundle_state_failure_still_scans_files(self)
# iteration will see a version mismatch and re-refresh rather than skip incorrectly
assert "mock_bundle" not in manager._bundle_versions
+ def test_refresh_dag_bundles_discovery_failure_keeps_known_files_and_dags(self):
Review Comment:
was solved by the `test_refresh_dag_bundles_discovery_failure_lists_again_on_next_refresh` method added when solving above comment about ordering between listing and version change.
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164730380
##########
airflow-core/tests/unit/utils/test_file.py:
##########
@@ -181,38 +180,6 @@ def test_get_modules_from_invalid_file(self):
assert len(modules) == 0
- def test_list_py_file_paths(self, test_zip_path):
Review Comment:
Added the test for `. airflowignore` and fixed the namings.
--
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] Discover Dag definitions through the Task SDK importers [airflow]
Posted by "dilnazanlid (via GitHub)" <gi...@apache.org>.
dilnazanlid commented on code in PR #74020:
URL: https://github.com/apache/airflow/pull/74020#discussion_r4164446264
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -970,56 +979,43 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative paths for dag files from bundle dir."""
- # Build up a list of Python files that could contain DAGs
- self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- rel_paths = [
- Path(x).relative_to(bundle.path)
- for x in list_py_file_paths(bundle.path, safe_mode=self.dag_discovery_safe_mode)
- ]
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> set[DagFileInfo]:
Review Comment:
Added one liner about the #74029 in the PR description - overall I think TP thought the discovery was forgotten before creating #74029, but I just divided it into 2 parts (wiring + this discovery).
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -950,10 +957,12 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._bundle_versions[bundle.name] = version_after_refresh
self._bundle_version_data[bundle.name] = version_data_after_refresh
- found_files = {
- DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path)
- for p in self._find_files_in_bundle(bundle)
- }
+ try:
+ found_files = self._find_files_in_bundle(bundle)
+ except Exception:
+ # Treating a failed listing as an empty bundle would deactivate all of its Dags.
Review Comment:
Replaced the comment in the file and PR description with new behavior.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -950,10 +957,12 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
self._bundle_versions[bundle.name] = version_after_refresh
self._bundle_version_data[bundle.name] = version_data_after_refresh
- found_files = {
- DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path)
- for p in self._find_files_in_bundle(bundle)
- }
+ try:
+ found_files = self._find_files_in_bundle(bundle)
Review Comment:
Good catch, moved the listing before the version advance and now it only runs if listing succeeds + added test case for it.
##########
airflow-core/src/airflow/dag_processing/dagbag.py:
##########
@@ -276,9 +276,7 @@ def get_dag(self, dag_id, *, session: Session = NEW_SESSION):
self.dags.pop(dag_id, None)
if dag is None or is_expired:
# Reprocess source file.
- found_dags = self.process_file(
- filepath=correct_maybe_zipped(orm_dag.fileloc), only_if_updated=False
- )
+ found_dags = self.process_file(filepath=orm_dag.fileloc, only_if_updated=False)
Review Comment:
I modified `test_refresh_packaged_dag` now and it adds a second dag member and asserts that only the expired dag's member is re-imported. It fails if correct_maybe_zipped is restored.
--
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