potiuk commented on code in PR #72336:
URL: https://github.com/apache/airflow/pull/72336#discussion_r3977695191
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1080,3 +1154,111 @@ async def get_mod_time(self, path: str) -> str: #
type: ignore[return]
return mod_time
except asyncssh.SFTPNoSuchFile:
raise AirflowException("No files matching")
+
+ async def sense_files_by_pattern(
+ self,
+ path: str,
+ fnmatch_pattern: str,
+ newer_than: datetime.datetime | None = None,
+ ) -> list[str]:
+ """
+ Return the names of files at ``path`` matching ``fnmatch_pattern``.
+
+ If ``newer_than`` is provided, only files modified after that
timestamp are returned; files
+ without a reported modification time are skipped in that case.
+
+ :param path: directory on the SFTP server to search for files matching
the pattern
+ :param fnmatch_pattern: pattern used to match filenames, see the
``fnmatch`` std library module
+ :param newer_than: if provided, only files modified after this UTC
timestamp are returned
+ """
+ files = await self.get_files_and_attrs_by_pattern(path=path,
fnmatch_pattern=fnmatch_pattern)
+ if not newer_than:
+ return [str(file.filename) for file in files]
+
+ matched_files = []
+ for file in files:
+ if file.attrs.mtime is None:
+ continue
+ if newer_than <= self._mod_time_to_utc(file.attrs.mtime):
+ matched_files.append(str(file.filename))
+ return matched_files
+
+ async def sense_path(self, path: str, newer_than: datetime.datetime | None
= None) -> bool:
+ """
+ Return whether ``path`` exists and, if ``newer_than`` is provided, was
modified since.
+
+ :param path: full path to the remote file
+ :param newer_than: if provided, the file must have been modified after
this UTC timestamp
+ """
+ mod_time = await self.get_mod_time(path)
+ if not newer_than:
+ return True
+ return newer_than <= self._mod_time_to_utc(mod_time)
+
+ @staticmethod
+ def _mod_time_to_utc(mod_time: int | float | str) -> datetime.datetime:
+ """Convert a modification time, either an epoch timestamp or
``%Y%m%d%H%M%S`` string, to UTC."""
+ if not isinstance(mod_time, str):
+ mod_time =
datetime.datetime.fromtimestamp(float(mod_time)).strftime("%Y%m%d%H%M%S")
+ return timezone.convert_to_utc(datetime.datetime.strptime(mod_time,
"%Y%m%d%H%M%S"))
+
+ async def transfer(
+ self,
+ operation: str,
+ local_filepath: str | list[str] | None,
+ remote_filepath: str | list[str],
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ """Perform an SFTP transfer operation (GET, PUT, or DELETE) using
native async I/O."""
+ if isinstance(local_filepath, str):
+ local_filepath_array = [local_filepath] if local_filepath else []
+ else:
+ local_filepath_array = local_filepath or []
+
+ if isinstance(remote_filepath, str):
+ remote_filepath_array = [remote_filepath]
+ else:
+ remote_filepath_array = list(remote_filepath)
+
+ semaphore = asyncio.Semaphore(concurrency)
+
+ async def _bounded(coro):
+ async with semaphore:
+ return await coro
+
+ async with await self._get_conn() as ssh_conn:
Review Comment:
This connection and its SFTP client are opened for every transfer, but only
the DELETE branch below actually uses `sftp` (`await sftp.unlink(remote)`).
The GET and PUT branches call `self.retrieve_file(...)` and
`self.store_file(...)`, and both of those open their own connection internally
— `retrieve_file` at line 1117 and `store_file` at line 1147 each start with
their own `async with await self._get_conn() as ssh_conn: async with
ssh_conn.start_sftp_client() as sftp:`.
So a deferred GET or PUT holds one idle SSH connection plus SFTP client open
for the entire duration of the transfer, on top of the real connections the
per-file calls make. That lands in the Triggerer, which is a shared process
running many triggers at once and where connection count is exactly the
resource you don't want to double.
Worth picking one of the two models rather than half of each: either open
the connection here once and have the branches use `sftp` directly (which also
gets you connection reuse across the files in a multi-path transfer, and would
be a genuine win over the sync path), or drop the outer `async with` entirely
and let `retrieve_file` / `store_file` manage their own. The first is better;
the second is a one-line change if you'd rather keep this PR's scope tight.
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1080,3 +1154,111 @@ async def get_mod_time(self, path: str) -> str: #
type: ignore[return]
return mod_time
except asyncssh.SFTPNoSuchFile:
raise AirflowException("No files matching")
+
+ async def sense_files_by_pattern(
+ self,
+ path: str,
+ fnmatch_pattern: str,
+ newer_than: datetime.datetime | None = None,
+ ) -> list[str]:
+ """
+ Return the names of files at ``path`` matching ``fnmatch_pattern``.
+
+ If ``newer_than`` is provided, only files modified after that
timestamp are returned; files
+ without a reported modification time are skipped in that case.
+
+ :param path: directory on the SFTP server to search for files matching
the pattern
+ :param fnmatch_pattern: pattern used to match filenames, see the
``fnmatch`` std library module
+ :param newer_than: if provided, only files modified after this UTC
timestamp are returned
+ """
+ files = await self.get_files_and_attrs_by_pattern(path=path,
fnmatch_pattern=fnmatch_pattern)
+ if not newer_than:
+ return [str(file.filename) for file in files]
+
+ matched_files = []
+ for file in files:
+ if file.attrs.mtime is None:
+ continue
+ if newer_than <= self._mod_time_to_utc(file.attrs.mtime):
+ matched_files.append(str(file.filename))
+ return matched_files
+
+ async def sense_path(self, path: str, newer_than: datetime.datetime | None
= None) -> bool:
+ """
+ Return whether ``path`` exists and, if ``newer_than`` is provided, was
modified since.
+
+ :param path: full path to the remote file
+ :param newer_than: if provided, the file must have been modified after
this UTC timestamp
+ """
+ mod_time = await self.get_mod_time(path)
+ if not newer_than:
+ return True
+ return newer_than <= self._mod_time_to_utc(mod_time)
+
+ @staticmethod
+ def _mod_time_to_utc(mod_time: int | float | str) -> datetime.datetime:
+ """Convert a modification time, either an epoch timestamp or
``%Y%m%d%H%M%S`` string, to UTC."""
+ if not isinstance(mod_time, str):
+ mod_time =
datetime.datetime.fromtimestamp(float(mod_time)).strftime("%Y%m%d%H%M%S")
+ return timezone.convert_to_utc(datetime.datetime.strptime(mod_time,
"%Y%m%d%H%M%S"))
+
+ async def transfer(
+ self,
+ operation: str,
+ local_filepath: str | list[str] | None,
+ remote_filepath: str | list[str],
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ """Perform an SFTP transfer operation (GET, PUT, or DELETE) using
native async I/O."""
+ if isinstance(local_filepath, str):
+ local_filepath_array = [local_filepath] if local_filepath else []
+ else:
+ local_filepath_array = local_filepath or []
+
+ if isinstance(remote_filepath, str):
+ remote_filepath_array = [remote_filepath]
+ else:
+ remote_filepath_array = list(remote_filepath)
+
+ semaphore = asyncio.Semaphore(concurrency)
+
+ async def _bounded(coro):
+ async with semaphore:
+ return await coro
+
+ async with await self._get_conn() as ssh_conn:
+ async with ssh_conn.start_sftp_client() as sftp:
+ if operation.lower() == SFTPOperation.GET:
+
+ async def _get(local: str, remote: str):
Review Comment:
This is where the async path diverges from the sync one it stands in for.
`_get` always calls `retrieve_file`, and `_put` always calls `store_file` —
there is no `isdir` branch, so a directory path that works with
`deferrable=False` fails with `deferrable=True`. The same applies to `_delete`,
which calls `sftp.unlink` and so fails on a directory that the sync path would
remove via `delete_directory(include_files=True)`.
Two more differences in the same block:
- **`prefetch` and `confirm` are accepted and silently ignored.**
`retrieve_file` is called without `prefetch`, `store_file` without `confirm`.
Silently dropping a parameter the user set is worse than not accepting it.
- **`create_intermediate_dirs` behaves differently.**
`os.makedirs(os.path.dirname(local), exist_ok=True)` raises `FileNotFoundError`
when `local` is a bare filename, because `os.path.dirname("f.txt")` is `""`.
The sync path does `Path(os.path.dirname(local)).mkdir(parents=True,
exist_ok=True)`, and `Path("")` is `.`, so it succeeds.
And the one I'd fix first — **DELETE of a missing file now raises in
deferrable mode**, where the sync path warns and skips. That silently regresses
#62639 for anyone who sets `deferrable=True`.
I realise the docstring positions deferrable mode as "best suited for single
large file transfers". But `concurrency`, `prefetch` and directory paths are
all still accepted here, so nothing stops a user from hitting these. Either
bring the async branch up to parity, or validate and reject the unsupported
combinations in `execute()` before deferring, so the user gets a clear error
instead of a silent behaviour change.
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1080,3 +1154,111 @@ async def get_mod_time(self, path: str) -> str: #
type: ignore[return]
return mod_time
except asyncssh.SFTPNoSuchFile:
raise AirflowException("No files matching")
+
+ async def sense_files_by_pattern(
+ self,
+ path: str,
+ fnmatch_pattern: str,
+ newer_than: datetime.datetime | None = None,
+ ) -> list[str]:
+ """
+ Return the names of files at ``path`` matching ``fnmatch_pattern``.
+
+ If ``newer_than`` is provided, only files modified after that
timestamp are returned; files
+ without a reported modification time are skipped in that case.
+
+ :param path: directory on the SFTP server to search for files matching
the pattern
+ :param fnmatch_pattern: pattern used to match filenames, see the
``fnmatch`` std library module
+ :param newer_than: if provided, only files modified after this UTC
timestamp are returned
+ """
+ files = await self.get_files_and_attrs_by_pattern(path=path,
fnmatch_pattern=fnmatch_pattern)
+ if not newer_than:
+ return [str(file.filename) for file in files]
+
+ matched_files = []
+ for file in files:
+ if file.attrs.mtime is None:
+ continue
+ if newer_than <= self._mod_time_to_utc(file.attrs.mtime):
+ matched_files.append(str(file.filename))
+ return matched_files
+
+ async def sense_path(self, path: str, newer_than: datetime.datetime | None
= None) -> bool:
+ """
+ Return whether ``path`` exists and, if ``newer_than`` is provided, was
modified since.
+
+ :param path: full path to the remote file
+ :param newer_than: if provided, the file must have been modified after
this UTC timestamp
+ """
+ mod_time = await self.get_mod_time(path)
+ if not newer_than:
+ return True
+ return newer_than <= self._mod_time_to_utc(mod_time)
+
+ @staticmethod
+ def _mod_time_to_utc(mod_time: int | float | str) -> datetime.datetime:
+ """Convert a modification time, either an epoch timestamp or
``%Y%m%d%H%M%S`` string, to UTC."""
+ if not isinstance(mod_time, str):
+ mod_time =
datetime.datetime.fromtimestamp(float(mod_time)).strftime("%Y%m%d%H%M%S")
+ return timezone.convert_to_utc(datetime.datetime.strptime(mod_time,
"%Y%m%d%H%M%S"))
+
+ async def transfer(
Review Comment:
This method is the feature — it's what actually runs in the Triggerer when
`deferrable=True` — and nothing in the PR exercises it.
`test_run_success` and `test_run_error` both patch `SFTPHookAsync.transfer`
wholesale, so they only prove the trigger wraps a result in a `TriggerEvent`.
The operator tests patch `SFTPHook.transfer`. The hook test file adds six
lines, and they assert the values of the `SFTPOperation` enum. Net coverage of
the ~55 lines here: none.
From the testing standards: *"Target exactly 100% coverage of what the PR
changes — no more, no less. Every changed or added behaviour must have a test;
every test must fail without the PR's change."*
This isn't a box-ticking request — both of the behavioural findings on this
PR are things a test would have caught. A test that asserts a directory GET
produces the same local tree in both modes would fail today. So would one that
deletes a missing file with `deferrable=True` and expects a warning, matching
`test_delete_missing_file_warns`. Mocking `asyncssh` at the `_get_conn`
boundary and asserting which hook methods get called for each operation would
also have made the duplicate-connection issue visible.
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -735,6 +741,74 @@ def get_files_by_pattern(self, path, fnmatch_pattern) ->
list[str]:
return matched_files
+ def transfer(
+ self,
+ operation: str,
+ local_filepath: str | list[str] | None,
+ remote_filepath: str | list[str],
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ """
+ Perform a synchronous SFTP transfer operation (GET, PUT, or DELETE).
+
+ Centralizes transfer logic so both the operator and the trigger
+ can delegate to the hook, in line with the DRY principle.
+
+ :param operation: The SFTP operation - put, get, or delete.
+ :param local_filepath: Local file path(s).
+ :param remote_filepath: Remote file path(s).
+ :param confirm: Whether to confirm file size after PUT (default: True).
+ :param create_intermediate_dirs: Create missing intermediate
directories (default: False).
+ :param concurrency: Number of threads for directory transfers
(default: 1).
+ :param prefetch: Whether to prefetch during GET (default: True).
+ """
+ if isinstance(local_filepath, str):
+ local_filepath_array = [local_filepath] if local_filepath else []
+ else:
+ local_filepath_array = local_filepath or []
+
+ if isinstance(remote_filepath, str):
+ remote_filepath_array = [remote_filepath]
+ else:
+ remote_filepath_array = list(remote_filepath)
+
+ if operation.lower() == SFTPOperation.GET:
+ for local, remote in zip(local_filepath_array,
remote_filepath_array):
+ if create_intermediate_dirs:
+ Path(os.path.dirname(local)).mkdir(parents=True,
exist_ok=True)
+ if self.isdir(remote):
+ if concurrency > 1:
+ self.retrieve_directory_concurrently(
+ remote, local, workers=concurrency,
prefetch=prefetch
+ )
+ else:
+ self.retrieve_directory(remote, local)
+ else:
+ self.retrieve_file(remote, local, prefetch=prefetch)
+ elif operation.lower() == SFTPOperation.PUT:
+ for local, remote in zip(local_filepath_array,
remote_filepath_array):
+ if create_intermediate_dirs:
+ self.create_directory(os.path.dirname(remote))
+ if os.path.isdir(local):
+ if concurrency > 1:
+ self.store_directory_concurrently(remote, local,
confirm=confirm, workers=concurrency)
+ else:
+ self.store_directory(remote, local, confirm=confirm)
+ else:
+ self.store_file(remote, local, confirm=confirm)
+ elif operation.lower() == SFTPOperation.DELETE:
+ for remote in remote_filepath_array:
+ if self.isdir(remote):
+ self.delete_directory(remote, include_files=True)
+ else:
+ try:
+ self.delete_file(remote)
+ except FileNotFoundError:
Review Comment:
The operator previously routed this through `_is_missing_path_error`, which
treated three shapes as "already absent": a `FileNotFoundError`, an `OSError`
whose `errno` is `ENOENT`, and an exception whose `args[0]` is `ENOENT`. That
helper is deleted here and replaced by this bare `except FileNotFoundError`.
The existing tests don't distinguish the two:
`test_delete_missing_file_warns` raises `FileNotFoundError("missing")` and
`test_delete_permission_error_raises` raises `PermissionError`, so both still
pass. That's exactly why it's worth asking rather than assuming.
The third branch in particular looks like it was written for an exception
that carries `ENOENT` in `args[0]` without being a `FileNotFoundError` —
paramiko does construct `IOError(errno.ENOENT, text)`, which Python maps to
`FileNotFoundError`, but that helper was added deliberately in #62639 to fix a
reported bug, and the extra branches suggest the reporter hit a shape the
simple check misses.
Could you confirm the narrowing is intentional? If it is, a test raising the
shape the old third branch covered — asserting it still warns — would lock the
decision down. If it isn't, keeping the helper (moved onto the hook alongside
the rest of the transfer logic) preserves the fix.
##########
providers/sftp/src/airflow/providers/sftp/operators/sftp.py:
##########
@@ -136,106 +139,85 @@ def execute(self, context: Any) -> str | list[str] |
None:
if self.operation.lower() not in (SFTPOperation.GET,
SFTPOperation.PUT, SFTPOperation.DELETE):
raise TypeError(
f"Unsupported operation value {self.operation}, "
- f"expected {SFTPOperation.GET} or {SFTPOperation.PUT} or
{SFTPOperation.DELETE}."
+ f"expected {SFTPOperation.GET!r}, {SFTPOperation.PUT!r}, "
+ f"or {SFTPOperation.DELETE!r}."
)
if self.concurrency < 1:
- raise ValueError(f"concurrency should be greater than 0, got
{self.concurrency}")
+ raise ValueError(f"concurrency should be >= 1, got
{self.concurrency}")
- file_msg = None
- try:
- if self.remote_host is not None:
- self.log.info(
- "remote_host is provided explicitly. "
- "It will replace the remote_host which was defined "
- "in sftp_hook or predefined in connection of ssh_conn_id."
+ # ------------------------------------------------------------------ #
+ # Synchronous path — delegate all transfer logic to the hook #
+ # ------------------------------------------------------------------ #
+ if self.remote_host is not None:
+ self.log.info(
+ "remote_host is provided explicitly. "
+ "It will replace the remote_host which was defined "
+ "in sftp_hook or predefined in connection of ssh_conn_id."
+ )
+
+ if self.ssh_conn_id:
+ if self.sftp_hook and isinstance(self.sftp_hook, SFTPHook):
+ self.log.info("ssh_conn_id is ignored when sftp_hook is
provided.")
+ else:
+ self.log.info("sftp_hook not provided or invalid. Trying
ssh_conn_id to create SFTPHook.")
+ self.sftp_hook = SFTPHook(
+ ssh_conn_id=self.ssh_conn_id,
+ remote_host=self.remote_host or "",
)
- if self.ssh_conn_id:
- if self.sftp_hook and isinstance(self.sftp_hook, SFTPHook):
- self.log.info("ssh_conn_id is ignored when sftp_hook is
provided.")
- else:
- self.log.info("sftp_hook not provided or invalid. Trying
ssh_conn_id to create SFTPHook.")
- self.sftp_hook = SFTPHook(
- ssh_conn_id=self.ssh_conn_id,
remote_host=self.remote_host or ""
- )
-
- if not self.sftp_hook:
- raise AirflowException("Cannot operate without sftp_hook or
ssh_conn_id.")
-
- if self.operation.lower() in (SFTPOperation.GET,
SFTPOperation.PUT):
- for _local_filepath, _remote_filepath in
zip(local_filepath_array, remote_filepath_array):
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(_local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True,
exist_ok=True)
- file_msg = f"from {_remote_filepath} to
{_local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- if self.sftp_hook.isdir(_remote_filepath):
- if self.concurrency > 1:
- self.sftp_hook.retrieve_directory_concurrently(
- _remote_filepath,
- _local_filepath,
- workers=self.concurrency,
- prefetch=self.prefetch,
- )
- elif self.concurrency == 1:
-
self.sftp_hook.retrieve_directory(_remote_filepath, _local_filepath)
- else:
- self.sftp_hook.retrieve_file(_remote_filepath,
_local_filepath)
- elif self.operation.lower() == SFTPOperation.PUT:
- remote_folder = os.path.dirname(_remote_filepath)
- if self.create_intermediate_dirs:
- self.sftp_hook.create_directory(remote_folder)
- file_msg = f"from {_local_filepath} to
{_remote_filepath}"
- self.log.info("Starting to transfer file %s", file_msg)
- if os.path.isdir(_local_filepath):
- if self.concurrency > 1:
- self.sftp_hook.store_directory_concurrently(
- _remote_filepath,
- _local_filepath,
- confirm=self.confirm,
- workers=self.concurrency,
- )
- elif self.concurrency == 1:
- self.sftp_hook.store_directory(
- _remote_filepath, _local_filepath,
confirm=self.confirm
- )
- else:
- self.sftp_hook.store_file(_remote_filepath,
_local_filepath, confirm=self.confirm)
- elif self.operation.lower() == SFTPOperation.DELETE:
- for _remote_filepath in remote_filepath_array:
- file_msg = f"{_remote_filepath}"
- self.log.info("Starting to delete %s", file_msg)
- try:
- if self.sftp_hook.isdir(_remote_filepath):
- self.sftp_hook.delete_directory(_remote_filepath,
include_files=True)
- else:
- self.sftp_hook.delete_file(_remote_filepath)
- except OSError as exc:
- if self._is_missing_path_error(exc):
- self.log.warning(
- "Remote path %s does not exist. Skipping
delete.", _remote_filepath
- )
- continue
- raise
+ if not self.sftp_hook:
+ raise AirflowException("Cannot operate without sftp_hook or
ssh_conn_id.")
+
+ if self.deferrable:
+ from airflow.providers.sftp.triggers.sftp import
SFTPTransferTrigger
+
+ self.defer(
+ trigger=SFTPTransferTrigger(
+ sftp_conn_id=self.ssh_conn_id or
SFTPHookAsync.default_conn_name,
Review Comment:
`SFTPHookAsync.default_conn_name` is `"sftp_default"`, so when `ssh_conn_id`
is unset this silently connects to a different server than the synchronous path
would.
The case that breaks is a Dag that supplies the hook rather than the conn id:
```python
SFTPOperator(
task_id="upload",
sftp_hook=SFTPHook(ssh_conn_id="my_prod_sftp"),
local_filepath="/tmp/data.csv",
remote_filepath="/incoming/data.csv",
)
```
That works today. Add `deferrable=True` and the transfer goes to
`sftp_default` instead — wrong host, no warning, and the task still reports
success. Note the check three lines above has already established
`self.sftp_hook` is set at this point, so the hook is known to exist and is
then discarded.
Both documented ways of configuring the operator need to survive the switch.
The hook's connection id is available as `self.sftp_hook.ssh_conn_id`, so
something along the lines of `self.ssh_conn_id or self.sftp_hook.ssh_conn_id`
would cover it — and if a case genuinely cannot be expressed for the async
hook, better to raise at `execute()` time than to fall back to a default
connection.
--
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]
