dilnazanlid commented on code in PR #72442:
URL: https://github.com/apache/airflow/pull/72442#discussion_r3934718187
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -958,12 +958,10 @@ def _refresh_dag_bundles(self, known_files: dict[str,
set[DagFileInfo]]):
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)
- ]
+ importer_registry = bundle.importer_registry
+ dag_files = importer_registry.list_dag_files(bundle.path,
safe_mode=self.dag_discovery_safe_mode)
Review Comment:
Good catch!
Currently, explicit `extensions` in the configuration only register entries
in `DagImporterRegistry._importers` (affecting `get_importer()` and
`can_handle()`), but `registry.list_dag_files()` delegates directly to
`importer.list_dag_files()`.
Because `AbstractDagImporter.list_dag_files()` inspects
`self.supported_extensions()` (which is a `@classmethod` returning class-level
defaults), files with custom configured extensions like `dag.custom` are
filtered out during directory discovery. Consequently, they are never enqueued
by `DagFileProcessorManager` and never reach parsing during normal bundle
execution.
Proposed fix:
- `DagImporterRegistry.list_dag_files()` performs a single walk of the
bundle directory.
- Each file is checked against `registry.get_importer(file_path)`. If
matched, content validation / safe mode is delegated to
`importer.might_contain_dag(file_path, safe_mode)`.
- We associate the configured extensions(global or bundle-level) with the
importer instance (`importer.set_configured_extensions()`) so that
`importer.can_handle()` and `importer.supported_extensions` stay consistent
with the registry.
This will fix file discovery for custom extensions, avoids running N
redundant filesystem walks across multiple importers, and strictly enforces
extension override precedence.
WDYT? I will implement it for this change into the
`dag_processing/importers` base classes. However, the importers are moved into
the SDK in https://github.com/apache/airflow/pull/72369 along with
DagDefinition abstraction replacing the filesystem path approach, so it will
also be reflected there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
