aoelvp94 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4478038743
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -140,6 +179,88 @@ def refresh(self) -> None:
delete_stale=True,
)
+ def _archive_state_dir(self) -> Path:
+ return self.s3_dags_dir.parent / f".{self.name}"
+
+ def _archive_etag_path(self) -> Path:
+ return self._archive_state_dir() / "etag"
+
+ def _remove_stale_archive_staging_dirs(self) -> None:
+ state_dir = self._archive_state_dir()
+ if not state_dir.is_dir():
+ return
+ for leftover in state_dir.iterdir():
+ if leftover.is_dir() and
leftover.name.startswith("archive-staging-"):
+ shutil.rmtree(leftover, ignore_errors=True)
+
+ def _download_archive(self, client, archive_path: Path, etag: str) -> str:
+ """Download the archive object named by ``etag``."""
+ request = {"Bucket": self.bucket_name, "Key": self.archive_key}
+ if etag:
+ request["IfMatch"] = etag
+ response = client.get_object(**request)
+ body = response["Body"]
+ try:
+ with archive_path.open("wb") as handle:
+ for chunk in body.iter_chunks():
+ handle.write(chunk)
+ finally:
+ body.close()
+ return response.get("ETag", "")
+
+ def _refresh_from_archive(self) -> None:
+ """Stage the Dag bundle by downloading and unpacking the single
archive object."""
+ self._remove_stale_archive_staging_dirs()
+ client = self.s3_hook.get_conn()
+ head = client.head_object(Bucket=self.bucket_name,
Key=self.archive_key)
+ etag: str = head.get("ETag", "")
+
+ marker = self._archive_etag_path()
+ if etag and marker.is_file() and marker.read_text() == etag:
Review Comment:
Good catch — that's a real bug. The marker lives beside the Dag folder, so
after a cleared tree `_initialize` recreated an empty directory, the ETag still
matched, and we skipped staging.
Done in
https://github.com/apache/airflow/pull/73015/commits/e9ed15391cd0e9f198d50c61f0d37aaf50c38d9a.
We now restage unless the staged tree actually has files, and the prefix path
drops the marker so switching back to an unchanged archive restages instead of
keeping the prefix sync's files. Added a test for the delete-and-reinitialize
case (bundle dir gone, matching marker still present → GET/unpack, not skip),
plus one for prefix-then-archive.
--
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]
