aoelvp94 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4407459280
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -140,6 +170,60 @@ def refresh(self) -> None:
delete_stale=True,
)
+ def _refresh_from_archive(self) -> None:
+ """Stage the Dag bundle by downloading and unpacking the single
archive object."""
+ 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.s3_dags_dir / self.archive_etag_marker
+ if etag and marker.is_file() and marker.read_text() == etag:
+ self._log.debug(
+ "Dag bundle archive 's3://%s/%s' is unchanged (ETag %s),
skipping staging",
+ self.bucket_name,
+ self.archive_key,
+ etag,
+ )
+ return
+
+ staging_dir = Path(tempfile.mkdtemp(dir=self.s3_dags_dir.parent,
prefix=".s3-archive-staging-"))
+ try:
+ archive_path = staging_dir / "_bundle_archive"
+ client.download_file(self.bucket_name, self.archive_key,
os.fspath(archive_path))
Review Comment:
Done in
[#73015](https://github.com/apache/airflow/pull/73015/commits/ad07c1b2d0162801dcc30fadf028ce95fb8b6ee1).
The archive is fetched with get_object and IfMatch set to the ETag from the
HEAD. The body is streamed, and the marker stores the ETag from that response,
or the HEAD ETag when the response has none. A replace between the HEAD and the
GET returns 412, the previous tree stays in place, and the staging directory is
removed. An empty ETag omits IfMatch. An unchanged archive is still one HEAD.
--
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]
