aoelvp94 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4053862005
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -136,6 +179,54 @@ 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))
+
+ unpack_dir = staging_dir / "unpacked"
+ unpack_dir.mkdir()
+ with tarfile.open(archive_path, "r:*") as tar:
+ tar.extractall(unpack_dir, filter="data")
+ archive_path.unlink()
+
+ if etag:
+ (unpack_dir / self.archive_etag_marker).write_text(etag)
+
+ # Swap the freshly unpacked tree into place so a partially staged
+ # bundle is never observable at self.s3_dags_dir.
+ old_dir = self.s3_dags_dir.parent /
f".s3-archive-old-{uuid.uuid4().hex}"
Review Comment:
https://github.com/apache/airflow/pull/73015/changes/3bbbd6d17e02ad20b0221de80546dd520a4e1297
pushed here, check please :)
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -126,6 +155,20 @@ def refresh(self) -> None:
raise AirflowException("Refreshing a specific version is not
supported")
with self.lock():
+ if self.archive_key:
+ try:
+ self._refresh_from_archive()
+ return
+ except Exception:
Review Comment:
https://github.com/apache/airflow/pull/73015/changes/3bbbd6d17e02ad20b0221de80546dd520a4e1297
pushed here, check please :)
--
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]
