aoelvp94 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4522097879
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -27,46 +30,59 @@
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``."""
+ with tarfile.open(archive_path, "r:*") as tar:
+ tar.extractall(unpack_dir, filter="data")
+
+
+class _S3DagBundleBase(BaseDagBundle):
"""
+ S3 connection, local directory, and console URL shared by the S3 bundle
classes.
- supports_versioning = False
+ :meta private:
+ """
def __init__(
self,
*,
aws_conn_id: str = AwsBaseHook.default_conn_name,
bucket_name: str,
- prefix: str = "",
**kwargs,
) -> None:
super().__init__(**kwargs)
self.aws_conn_id = aws_conn_id
self.bucket_name = bucket_name
- self.prefix = prefix
# Local path where S3 Dags are downloaded
self.s3_dags_dir: Path = self.base_dir
-
- log = structlog.get_logger(__name__)
- self._log = log.bind(
- bundle_name=self.name,
- version=self.version,
- bucket_name=self.bucket_name,
- prefix=self.prefix,
- aws_conn_id=self.aws_conn_id,
- )
+ self._log = self._get_logger()
self._s3_hook: S3Hook | None = None
+ def _log_bindings(self) -> dict[str, str | None]:
+ """Return log context shared by every S3 Dag bundle."""
+ return {
+ "bundle_name": self.name,
+ "version": self.version,
+ "bucket_name": self.bucket_name,
+ "aws_conn_id": self.aws_conn_id,
+ }
+
+ def _get_logger(self):
Review Comment:
Added `-> FilteringBoundLogger` on the base and both subclasses (imported
under `TYPE_CHECKING`, as in `dag_processing/processor.py`).
https://github.com/aoelvp94/airflow/commit/401e75d05662542406ef2b82df176834506cab7e
--
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]
