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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
