aoelvp94 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4491292726
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -163,7 +150,215 @@ def view_url_template(self) -> str | None:
if self.s3_hook.region_name:
url += f".{self.s3_hook.region_name}"
url += ".amazonaws.com"
- if self.prefix:
- url += f"/{self.prefix}"
+ key = self._view_url_key()
+ if key:
+ url += f"/{key}"
return url
+
+ def _reject_version_refresh(self) -> None:
+ if self.version:
+ raise AirflowException("Refreshing a specific version is not
supported")
+
+
+class S3DagBundle(_S3DagBundleBase):
+ """
+ S3 Dag bundle - exposes a directory in S3 as a Dag bundle.
+
+ 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).
+ """
+
+ supports_versioning = False
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str = AwsBaseHook.default_conn_name,
+ bucket_name: str,
+ prefix: str = "",
+ **kwargs,
+ ) -> None:
+ self.prefix = prefix
+ super().__init__(aws_conn_id=aws_conn_id, bucket_name=bucket_name,
**kwargs)
+
+ def _log_fields(self) -> dict[str, str]:
+ return {"prefix": self.prefix}
+
+ def _view_url_key(self) -> str:
+ return self.prefix
+
+ def _validate_remote(self) -> None:
+ if self.prefix:
+ # don't check when prefix is ""
+ if not self.s3_hook.check_for_prefix(
+ bucket_name=self.bucket_name, prefix=self.prefix, delimiter="/"
+ ):
+ raise AirflowException(f"S3 prefix
's3://{self.bucket_name}/{self.prefix}' does not exist.")
+
+ def __repr__(self):
+ return (
+ f"<S3DagBundle("
+ f"name={self.name!r}, "
+ f"bucket_name={self.bucket_name!r}, "
+ f"prefix={self.prefix!r}, "
+ f"version={self.version!r}"
+ f")>"
+ )
+
+ def refresh(self) -> None:
+ """Refresh the Dag bundle by re-downloading the Dags from S3."""
+ self._reject_version_refresh()
+
+ with self.lock():
+ # Same bundle name may previously have been an S3ArchiveDagBundle.
Drop its
+ # ETag so switching back to the archive restages instead of
keeping these files.
+ _get_archive_etag_path(self.s3_dags_dir,
self.name).unlink(missing_ok=True)
Review Comment:
Moved the ETag marker into the staged tree (`.archive_etag`) so clearing or
replacing the tree drops it. `S3DagBundle` no longer touches archive state.
https://github.com/aoelvp94/airflow/commit/6bd086603f4fbb4a51ed773c697682e77bb0668c
--
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]
