ramitkataria commented on code in PR #73269:
URL: https://github.com/apache/airflow/pull/73269#discussion_r4077695771
##########
providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_ftp.py:
##########
@@ -109,16 +109,25 @@ def execute(self, context: Context):
self.log.info("Getting files in s3://%s/%s", self.s3_bucket,
self.s3_key)
all_keys = s3_hook.list_keys(bucket_name=self.s3_bucket,
prefix=self.s3_key) or []
filenames = [k[len(self.s3_key) :] for k in all_keys]
- if self.s3_filenames == "*":
+ s3_prefix: str = self.s3_filenames
+ if s3_prefix == "*":
files = filenames
else:
- s3_prefix: str = self.s3_filenames
- files = [f for f in filenames if s3_prefix in f]
+ files = [f for f in filenames if f.startswith(s3_prefix)]
Review Comment:
One thing that was left open from the earlier thread:
`list_keys(prefix=s3_key)` also returns nested keys, so a relative name like
`2024/pre_one.txt` no longer starts with the prefix and ends up in `dropped`.
The `"*"` branch in this same operator still transfers those nested keys, and
`FTPToS3Operator` would select them via the basename match, so the four
operators still disagree on layouts with subdirectories.
Could we either match on `posixpath.basename(f)` here and in
`S3ToSFTPOperator` (and do the rename on the basename too, so `replace(..., 1)`
cannot hit a directory segment), or state in the changelog that the S3-side
prefix applies to the full path relative to `s3_key`? I am fine with either, I
mostly want the behavior to be intentional and documented
##########
providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_ftp.py:
##########
@@ -109,16 +109,25 @@ def execute(self, context: Context):
self.log.info("Getting files in s3://%s/%s", self.s3_bucket,
self.s3_key)
all_keys = s3_hook.list_keys(bucket_name=self.s3_bucket,
prefix=self.s3_key) or []
filenames = [k[len(self.s3_key) :] for k in all_keys]
- if self.s3_filenames == "*":
+ s3_prefix: str = self.s3_filenames
+ if s3_prefix == "*":
files = filenames
else:
- s3_prefix: str = self.s3_filenames
- files = [f for f in filenames if s3_prefix in f]
+ files = [f for f in filenames if f.startswith(s3_prefix)]
+ dropped = [f for f in filenames if s3_prefix in f and not
f.startswith(s3_prefix)]
+ if dropped:
+ self.log.warning(
Review Comment:
Minor: this logs the full `dropped` list at WARNING on every run, and for a
bucket or directory where the non-prefix siblings are a permanent, legitimate
part of the layout it will keep doing so forever. With nested S3 keys, the list
can get long.
Could we log the count plus a bounded sample, say the first 10 entries? Same
in the other three operators
--
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]
