srchilukoori commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4276474036
Thanks for adding deferrable support to `SFTPOperator` — the overall
approach is sound. A few issues to address before this is merge-ready.
---
**[Blocker 1] Circular import forces inner imports** — `triggers/sftp.py`
lines 192, 210–214
`operators/sftp.py` imports `SFTPOperatorTrigger` at module level;
`_run_transfer` imports `SFTPOperation` back from `operators/sftp.py`, creating
a circular. The inner imports are a workaround but violate project coding
guidelines.
Fix: extract `SFTPOperation` into a `constants.py` module (it's just string
constants) and import it at the top of both files. `os`, `Path`, and `SFTPHook`
in `_run_transfer` can also move to module level.
---
**[Blocker 2] Missing type annotations on `SFTPOperatorTrigger`** —
`triggers/sftp.py` line 159
`SFTPTrigger` in the same file is fully annotated. Please be consistent:
```python
def __init__(
self,
ssh_conn_id: str | None = None,
local_filepath: str | list[str] | None = None,
remote_filepath: str | list[str] = "",
operation: str = SFTPOperation.PUT,
confirm: bool = True,
create_intermediate_dirs: bool = False,
remote_host: str | None = None, # see Blocker 3
concurrency: int = 1, # see Blocker 3
prefetch: bool = True, # see Blocker 3
) -> None:
```
Same for `serialize() -> tuple[str, dict[str, Any]]`, `run() ->
AsyncIterator[TriggerEvent]`, `_run_transfer() -> None`.
---
**[Blocker 3] `remote_host`, `concurrency`, `prefetch` silently dropped** —
`triggers/sftp.py` line 216
`execute()` uses all three but the trigger ignores them. Users get different
behavior in deferrable mode with no error.
```python
# _run_transfer — forward remote_host at minimum:
sftp_hook = SFTPHook(ssh_conn_id=self.ssh_conn_id,
remote_host=self.remote_host or "")
```
Also add these three params to `__init__`, `serialize()`, and the `defer()`
call in `SFTPOperator.execute()`.
---
**[Blocker 4] Directory transfers broken in deferrable mode** —
`triggers/sftp.py` line 208
`execute()` checks `sftp_hook.isdir()` and dispatches to
`retrieve_directory_concurrently` / `store_directory_concurrently`.
`_run_transfer` calls `retrieve_file`/`store_file` unconditionally — a
directory path raises `IOError` at runtime.
Suggested fix: extract the transfer loop from `execute()` into a
`@staticmethod _do_transfer(sftp_hook, ...)` and call it from both `execute()`
and `_run_transfer()`. Keeps both paths in sync without duplication.
---
**[Blocker 5] `asyncio.get_event_loop()` deprecated** — `triggers/sftp.py`
line 195
Deprecated since Python 3.10+. Use `get_running_loop()` inside a coroutine:
```python
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._run_transfer)
```
---
**[Minor 1] `serialize()` missing new params** — `triggers/sftp.py` line 176
`remote_host`, `concurrency`, `prefetch` must be included in `serialize()`'s
dict or the trigger can't be reconstructed after a triggerer restart.
---
**[Minor 2] Redundant imports inside test methods** — `test_sftp.py` lines
687, 706, 722–725
`pytest`, `SFTPOperation`, `SFTPOperator`, and `AirflowException` are
already imported at the top of the file. Please remove the inner imports.
---
**[Minor 3] Missing trigger tests** — `test_sftp.py`
Please add tests for `SFTPOperatorTrigger`:
- `serialize()` roundtrip — ensures trigger survives a triggerer restart
- `run()` success path: mock `_run_transfer`, assert
`TriggerEvent(status=success)`
- `run()` error path: mock `_run_transfer` raising, assert
`TriggerEvent(status=error)`
- GET and DELETE operations in `_run_transfer`
--
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]
