aoelvp94 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4488135282
##########
providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py:
##########
@@ -262,3 +282,285 @@ def test_refresh_uses_directory_prefix(self, s3_client,
prefix, extra_suffix):
assert repr(bundle) == original_repr
assert bundle.view_url_template() == original_url
assert s3_client.get_object(Bucket=S3_BUCKET_NAME,
Key=extra_key)["Body"].read() == b"outside"
+
+ def test_rejects_archive_key(self):
+ with pytest.raises(TypeError, match="archive_key"):
+ S3DagBundle(
+ name="test",
+ aws_conn_id=AWS_CONN_ID_DEFAULT,
+ bucket_name=S3_BUCKET_NAME,
+ prefix=S3_BUCKET_PREFIX,
+ archive_key=S3_ARCHIVE_KEY,
Review Comment:
Agreed — dropped both; they only exercise the `TypeError` Python already
raises.
https://github.com/aoelvp94/airflow/commit/054298f048098bf7385fe5bac213658311269ad9
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -27,33 +30,44 @@
from airflow.providers.common.compat.sdk import AirflowException
-class S3DagBundle(BaseDagBundle):
- """
- S3 Dag bundle - exposes a directory in S3 as a Dag bundle.
+def _require_tar_data_filter() -> None:
+ if hasattr(tarfile, "data_filter"):
+ return
+ raise RuntimeError("Archive Dag bundles require Python 3.11.4 or later
(tarfile filter='data').")
- This allows Airflow to load Dags directly from an S3 bucket.
- :param aws_conn_id: Airflow connection ID for AWS. Defaults to
AwsBaseHook.default_conn_name.
- :param bucket_name: The name of the S3 bucket containing the Dag files.
- :param prefix: Optional subdirectory within the S3 bucket where the Dags
are stored.
- A trailing slash is optional.
- If None, Dags are assumed to be at the root of the bucket
(Optional).
+def _extract_archive(archive_path: Path, unpack_dir: Path) -> None:
+ """Unpack a bundle archive without writing outside ``unpack_dir``."""
+ _require_tar_data_filter()
+ with tarfile.open(archive_path, "r:*") as tar:
+ tar.extractall(unpack_dir, filter="data")
+
+
+def _archive_state_dir(bundle_dir: Path, bundle_name: str) -> Path:
+ return bundle_dir.parent / f".{bundle_name}"
+
+
+def _archive_etag_path(bundle_dir: Path, bundle_name: str) -> Path:
+ return _archive_state_dir(bundle_dir, bundle_name) / "etag"
Review Comment:
Agreed — renamed to `_get_archive_state_dir` / `_get_archive_etag_path`.
https://github.com/aoelvp94/airflow/commit/0bbe8ae910c10ce041c2f3c84ee32ba0e9e1c798
--
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]
