You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "sunildataengineer (via GitHub)" <gi...@apache.org> on 2026/04/19 06:04:36 UTC
[PR] Add deferrable mode to SFTPOperator [airflow]
sunildataengineer opened a new pull request, #65480:
URL: https://github.com/apache/airflow/pull/65480
What this PR does
Adds `deferrable=True` parameter to `SFTPOperator` which allows the
operator to defer execution to `SFTPOperatorTrigger`, freeing the
worker slot during file transfers instead of blocking it.
Currently, `SFTPOperator` blocks a worker slot for the entire duration
of every file transfer. For large file transfers, this wastes worker
resources unnecessarily.
Changes
- Added `SFTPOperatorTrigger` class in `triggers/sftp.py`
- Added `deferrable` parameter to `SFTPOperator.__init__()` with
default `False` (fully backward compatible)
- Added `self.defer()` call in `execute()` when `deferrable=True`
- Added `execute_complete()` callback method to handle trigger result
- Added 3 unit tests covering deferrable mode
Tests
- `test_sftp_operator_defers_when_deferrable_true`
- `test_sftp_operator_execute_complete_success`
- `test_sftp_operator_execute_complete_raises_on_error`
Closes #65475
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4503371349
Hi @srchilukoori π Gentle bump on this β it's been just over a month since your review. I've addressed all 5 blockers and 3 minor issues you raised:
- Extracted `SFTPOperation` into `constants.py` (circular import fixed)
- Full type annotations added to `SFTPOperatorTrigger`
- remote_host`, `concurrency`, `prefetch` forwarded through trigger + `serialize()`
- Directory transfer logic extracted into `_do_transfer()` static method
- Replaced `asyncio.get_event_loop()` with `get_running_loop()`
- Inner imports removed from test methods
- Added `serialize()` roundtrip + `run()` tests for `SFTPOperatorTrigger`
Would love a re-review when you get a chance. Happy to make any further changes! π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3291406527
##########
providers/sftp/src/airflow/providers/sftp/operators/sftp.py:
##########
@@ -108,8 +105,25 @@ def __init__(
self.remote_filepath = remote_filepath
self.concurrency = concurrency
self.prefetch = prefetch
+ self.deferrable = deferrable
def execute(self, context: Any) -> str | list[str] | None:
+ if self.deferrable:
+ self.defer(
+ trigger=SFTPOperatorTrigger(
+ ssh_conn_id=self.ssh_conn_id,
+ local_filepath=self.local_filepath,
+ remote_filepath=self.remote_filepath,
+ operation=self.operation,
+ confirm=self.confirm,
+ create_intermediate_dirs=self.create_intermediate_dirs,
+ remote_host=self.remote_host,
+ concurrency=self.concurrency,
+ prefetch=self.prefetch,
+ ),
+ method_name="execute_complete",
+ )
+
if self.local_filepath is None:
Review Comment:
Good catch @dabla β you're right, `SFTPOperator.execute()` is not calling `SFTPHook.transfer()` yet. That was the whole point of the refactoring. Will fix it now so the operator fully delegates to the hook, and same for the trigger delegating to `SFTPHookAsync.transfer()`. Push incoming shortly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4517138885
@dabla β all 3 changes addressed:
1. serialize() now uses f"{self.__class__.__module__}.{self.__class__.__name__}"
2. Transfer logic moved to transfer() method on both SFTPHook and SFTPHookAsync
3. Trigger now uses SFTPHookAsync.transfer() β _do_transfer removed
Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3289030775
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -706,6 +706,75 @@ 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).
+
+ Centralises transfer logic so both the operator and the trigger
+ can delegate to the hook, in line with the DRY principle.
Review Comment:
Just change docstring to this, it is logic that we want to respect DRY principle, no need to mention it in a docstring of a method:
`Perform an SFTP transfer operation (GET, PUT, or DELETE) via a thread executor.`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4643157619
> @dabla @potiuk @srchilukoori @kaxil β fixed docs (asgiref added to requirements table) and added newsfragment 65480.feature.rst. Pushing now. π
200 changed files, I think something went wrong?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4531574335
@dabla β branch rebased, CI issues being resolved. All your feedback
has been addressed. Could you re-review when you get a chance? π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4365220000
@srchilukoori @potiuk
Rebased on latest upstream main. All 6 tests passing locally.
Ruff checks passing. Ready for re-review.
- Rebased on upstream/main (April 2026)
- All ruff linting issues resolved
- 6 tests passing: 3 operator tests + 3 trigger tests
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3289069898
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -139,3 +139,135 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperatorTrigger(BaseTrigger):
+ """
+ Trigger for SFTPOperator deferrable mode.
+
+ Fires when a file transfer (PUT, GET, or DELETE) completes
+ on the SFTP server, freeing the worker slot during the transfer.
+
+ :param ssh_conn_id: The SSH connection ID to use.
+ :param local_filepath: Local file path(s) to transfer.
+ :param remote_filepath: Remote file path(s) on the SFTP server.
+ :param operation: The SFTP operation - put, get, or delete.
+ :param confirm: Whether to confirm the file transfer.
+ :param create_intermediate_dirs: Whether to create intermediate dirs.
+ :param remote_host: Remote host to connect to (overrides connection).
+ :param concurrency: Number of threads for directory transfers.
+ :param prefetch: Whether to prefetch during file retrieval.
+ """
+
+ def __init__(
+ self,
+ ssh_conn_id: str | None = None,
+ local_filepath: str | list[str] | None = None,
+ remote_filepath: str | list[str] = "",
+ operation: str = "put",
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ remote_host: str | None = None,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ super().__init__()
+ self.ssh_conn_id = ssh_conn_id
+ self.local_filepath = local_filepath
+ self.remote_filepath = remote_filepath
+ self.operation = operation
+ self.confirm = confirm
+ self.create_intermediate_dirs = create_intermediate_dirs
+ self.remote_host = remote_host
+ self.concurrency = concurrency
+ self.prefetch = prefetch
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize the trigger for storage in the database."""
+ return (
+ f"{self.__class__.__module__}.{self.__class__.__name__}",
+ {
+ "ssh_conn_id": self.ssh_conn_id,
+ "local_filepath": self.local_filepath,
+ "remote_filepath": self.remote_filepath,
+ "operation": self.operation,
+ "confirm": self.confirm,
+ "create_intermediate_dirs": self.create_intermediate_dirs,
+ "remote_host": self.remote_host,
+ "concurrency": self.concurrency,
+ "prefetch": self.prefetch,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Run the file transfer asynchronously and yield a TriggerEvent when done."""
+ try:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(
Review Comment:
just call and await the transfer method from async SFTPHook
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4277998385
Thank you for the detailed review @srchilukoori.
The overall approach feedback is very encouraging. I will
address all the blockers and minor issues:
1. Extract SFTPOperation into constants.py to fix circular import
2. Add full type annotations to SFTPOperatorTrigger
3. Add remote_host, concurrency, prefetch params to trigger
4. Extract transfer logic into @staticmethod _do_transfer()
5. Replace asyncio.get_event_loop() with get_running_loop()
6. Add serialize() roundtrip and run() tests for trigger
7. Remove inner imports from test methods
Will push the updated code shortly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4662480182
@dabla @potiuk β the "200 changed files" in the GitHub Files tab
was showing all commits including upstream merge commits.
The actual diff vs apache:main is only 11 files:
providers/sftp/src/.../constants.py
providers/sftp/src/.../hooks/sftp.py
providers/sftp/src/.../operators/sftp.py
providers/sftp/src/.../triggers/sftp.py
providers/sftp/tests/.../test_sftp.py
providers/sftp/tests/.../test_constants.py
providers/sftp/pyproject.toml
providers/sftp/docs/index.rst
providers/sftp/newsfragments/65480.feature.rst
scripts/ci/prek/known_airflow_exceptions.txt
uv.lock
I'll squash the commits to make the PR cleaner. π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4575397194
> @dabla @potiuk β branch rebased on latest main. Static checks, docs, and unit test failures β could you point me to the specific failing check so I can fix it quickly? I want to get this across the finish line. π
There are still quite some errors present.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4275301634
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better π.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3289059991
##########
providers/sftp/src/airflow/providers/sftp/operators/sftp.py:
##########
@@ -108,8 +105,25 @@ def __init__(
self.remote_filepath = remote_filepath
self.concurrency = concurrency
self.prefetch = prefetch
+ self.deferrable = deferrable
def execute(self, context: Any) -> str | list[str] | None:
+ if self.deferrable:
+ self.defer(
+ trigger=SFTPOperatorTrigger(
+ ssh_conn_id=self.ssh_conn_id,
+ local_filepath=self.local_filepath,
+ remote_filepath=self.remote_filepath,
+ operation=self.operation,
+ confirm=self.confirm,
+ create_intermediate_dirs=self.create_intermediate_dirs,
+ remote_host=self.remote_host,
+ concurrency=self.concurrency,
+ prefetch=self.prefetch,
+ ),
+ method_name="execute_complete",
+ )
+
if self.local_filepath is None:
Review Comment:
Seems like the SFTPOperator is not calling the transfer method of the SFTPHook? That was the whole purpose of the refactoring.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4661892607
@dabla @potiuk β the "200 changed files" in the GitHub Files tab
was showing all commits including upstream merge commits.
The actual diff vs apache:main is only 11 files:
- providers/sftp/src/.../constants.py
- providers/sftp/src/.../hooks/sftp.py
- providers/sftp/src/.../operators/sftp.py
- providers/sftp/src/.../triggers/sftp.py
- providers/sftp/tests/.../test_sftp.py
- providers/sftp/tests/.../test_constants.py
- providers/sftp/pyproject.toml
- providers/sftp/docs/index.rst
- providers/sftp/newsfragments/65480.feature.rst
- scripts/ci/prek/known_airflow_exceptions.txt
- uv.lock
I'll squash the commits to make the PR cleaner. π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "srchilukoori (via GitHub)" <gi...@apache.org>.
srchilukoori commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4305161763
Good progress β all 5 original blockers are resolved. A few remaining issues are likely causing the CI failures, plus two consistency nits.
---
**[Blocker] Inner imports still in `_do_transfer`** β `triggers/sftp.py`
The circular import is gone now that `SFTPOperation` lives in `constants.py`, so there is no reason for `os`, `Path`, `SFTPOperation`, and `SFTPHook` to be imported inside the method. Static checks (`ruff`/import-order) will fail on this pattern. Move them to the top of the file:
```python
import os
from pathlib import Path
from airflow.providers.sftp.constants import SFTPOperation
from airflow.providers.sftp.hooks.sftp import SFTPHook
```
---
**[Blocker] Inner imports still in new test methods** β `test_sftp.py`
`test_sftp_operator_defers_when_deferrable_true` has an inner `from airflow.providers.sftp.triggers.sftp import SFTPOperatorTrigger` β `SFTPOperatorTrigger` is already imported at the top of the file.
All three `TestSFTPOperatorTrigger` methods (`test_serialize_roundtrip`, `test_run_success`, `test_run_error`) import `SFTPOperatorTrigger`, `asyncio`, and `unittest.mock.patch` inside the method body. Move all of these to the top of the file with the existing imports.
---
**[Fix] `remote_host or ""` passes empty string to SFTPHook** β `triggers/sftp.py`
```python
# current β wrong: overrides the connection's host with "" when remote_host is None
sftp_hook = SFTPHook(ssh_conn_id=self.ssh_conn_id, remote_host=self.remote_host or "")
# fix β pass None so the hook uses the connection's host
sftp_hook = SFTPHook(ssh_conn_id=self.ssh_conn_id, remote_host=self.remote_host)
```
---
**[Nit] Use `SFTPOperation.PUT` as default instead of raw string** β `triggers/sftp.py` line ~170
```python
# current
operation: str = "put",
# preferred β consistent with SFTPOperator
operation: str = SFTPOperation.PUT,
```
---
Once the inner imports are moved to module level the static checks should pass. Run `prek run --from-ref main --stage pre-commit` locally before pushing to catch any remaining lint issues.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4468031680
@potiuk @ashb @gopidesupavan β all 3 CI failures fixed:
- Added missing test_constants.py
- Fixed conf import to use airflow.providers.common.compat.sdk
- Updated AirflowException allowlist
Branch is rebased on main. Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4531573208
Hi @dabla π PR #65480 (deferrable SFTPOperator) β all your review
comments have been addressed including the rename to SFTPTrigger and
asgiref sync_to_async. Branch is rebased on main. Could you take a
final look when you get a chance? π
https://github.com/apache/airflow/pull/65480
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4525114134
@dabla β both addressed:
1. SFTPOperatorTrigger renamed to SFTPTrigger (consistent with other providers)
2. SFTPHookAsync.transfer() now uses asgiref sync_to_async instead of asyncio.run()
3. asgiref>=3.5.2 added to provider dependencies
Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3292352687
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -139,3 +139,135 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperatorTrigger(BaseTrigger):
Review Comment:
Missed this one, SFTPOperatorTrigger should in fact be called SFTPTrigger, to be consistent in naming convention like with other providers
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4544974755
@dabla β on it right now. Will push fixes shortly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3279964839
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -139,3 +139,135 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperatorTrigger(BaseTrigger):
+ """
+ Trigger for SFTPOperator deferrable mode.
+
+ Fires when a file transfer (PUT, GET, or DELETE) completes
+ on the SFTP server, freeing the worker slot during the transfer.
+
+ :param ssh_conn_id: The SSH connection ID to use.
+ :param local_filepath: Local file path(s) to transfer.
+ :param remote_filepath: Remote file path(s) on the SFTP server.
+ :param operation: The SFTP operation - put, get, or delete.
+ :param confirm: Whether to confirm the file transfer.
+ :param create_intermediate_dirs: Whether to create intermediate dirs.
+ :param remote_host: Remote host to connect to (overrides connection).
+ :param concurrency: Number of threads for directory transfers.
+ :param prefetch: Whether to prefetch during file retrieval.
+ """
+
+ def __init__(
+ self,
+ ssh_conn_id: str | None = None,
+ local_filepath: str | list[str] | None = None,
+ remote_filepath: str | list[str] = "",
+ operation: str = "put",
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ remote_host: str | None = None,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ super().__init__()
+ self.ssh_conn_id = ssh_conn_id
+ self.local_filepath = local_filepath
+ self.remote_filepath = remote_filepath
+ self.operation = operation
+ self.confirm = confirm
+ self.create_intermediate_dirs = create_intermediate_dirs
+ self.remote_host = remote_host
+ self.concurrency = concurrency
+ self.prefetch = prefetch
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize the trigger for storage in the database."""
+ return (
+ "airflow.providers.sftp.triggers.sftp.SFTPOperatorTrigger",
+ {
+ "ssh_conn_id": self.ssh_conn_id,
+ "local_filepath": self.local_filepath,
+ "remote_filepath": self.remote_filepath,
+ "operation": self.operation,
+ "confirm": self.confirm,
+ "create_intermediate_dirs": self.create_intermediate_dirs,
+ "remote_host": self.remote_host,
+ "concurrency": self.concurrency,
+ "prefetch": self.prefetch,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Run the file transfer asynchronously and yield a TriggerEvent when done."""
+ try:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(
+ None,
+ self._do_transfer,
+ )
+ yield TriggerEvent(
+ {
+ "status": "success",
+ "local_filepath": self.local_filepath,
+ }
+ )
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+
+ def _do_transfer(self) -> None:
+ """Run the actual synchronous SFTP transfer in a thread executor."""
+ import os
+ from pathlib import Path
+
+ from airflow.providers.sftp.constants import SFTPOperation
+ from airflow.providers.sftp.hooks.sftp import SFTPHook
+
+ sftp_hook = SFTPHook(
+ ssh_conn_id=self.ssh_conn_id,
+ remote_host=self.remote_host or "",
+ )
+
+ if isinstance(self.local_filepath, str):
+ local_filepath_array = [self.local_filepath] if self.local_filepath else []
+ else:
+ local_filepath_array = self.local_filepath or []
+
+ if isinstance(self.remote_filepath, str):
+ remote_filepath_array = [self.remote_filepath]
+ else:
+ remote_filepath_array = list(self.remote_filepath)
+
+ if self.operation.lower() == SFTPOperation.GET:
Review Comment:
Would it make sense to move the _do_transfer logic into the hook? We currently have duplicated logic in both the trigger and the operator (`execute` method).
Ideally, the `SFTPOperator` should delegate this responsibility to the hook so the logic lives in a single, reusable place, in line with the DRY principle.
So far weβve gotten away with this duplication because the logic was only implemented in the operatorβs execute method. However, even that was not ideal: operators should primarily act as lightweight wrappers for DAG usage, while the actual implementation logic should reside in hooks. This ensures that users relying directly on the hook have access to the same functionality without depending on the operator, and avoids divergence between execution paths.
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -139,3 +139,135 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperatorTrigger(BaseTrigger):
+ """
+ Trigger for SFTPOperator deferrable mode.
+
+ Fires when a file transfer (PUT, GET, or DELETE) completes
+ on the SFTP server, freeing the worker slot during the transfer.
+
+ :param ssh_conn_id: The SSH connection ID to use.
+ :param local_filepath: Local file path(s) to transfer.
+ :param remote_filepath: Remote file path(s) on the SFTP server.
+ :param operation: The SFTP operation - put, get, or delete.
+ :param confirm: Whether to confirm the file transfer.
+ :param create_intermediate_dirs: Whether to create intermediate dirs.
+ :param remote_host: Remote host to connect to (overrides connection).
+ :param concurrency: Number of threads for directory transfers.
+ :param prefetch: Whether to prefetch during file retrieval.
+ """
+
+ def __init__(
+ self,
+ ssh_conn_id: str | None = None,
+ local_filepath: str | list[str] | None = None,
+ remote_filepath: str | list[str] = "",
+ operation: str = "put",
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ remote_host: str | None = None,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ super().__init__()
+ self.ssh_conn_id = ssh_conn_id
+ self.local_filepath = local_filepath
+ self.remote_filepath = remote_filepath
+ self.operation = operation
+ self.confirm = confirm
+ self.create_intermediate_dirs = create_intermediate_dirs
+ self.remote_host = remote_host
+ self.concurrency = concurrency
+ self.prefetch = prefetch
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize the trigger for storage in the database."""
+ return (
+ "airflow.providers.sftp.triggers.sftp.SFTPOperatorTrigger",
Review Comment:
I would make generation of qualified class name of trigger dynamic, so it's refactor proof in the future if renaming would occur.
So instead of:
`"airflow.providers.sftp.triggers.sftp.SFTPOperatorTrigger",`
I would write it like:
`f"{self.__class__.__module__}.{self.__class__.__name__}",`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3292356389
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1040,3 +1109,30 @@ 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 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:
+ import asyncio
Review Comment:
Wondering if asgiref sync_to_async would nβt be a better solution?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer closed pull request #65480: Add deferrable mode to SFTPOperator
URL: https://github.com/apache/airflow/pull/65480
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4658959218
@sunildataengineer β A reviewer (@dabla) has requested changes on this PR, so I've removed the `ready for maintainer review` label β the next step is on your side. Could you address the review comments (push a fix, or reply in-thread explaining why the feedback doesn't apply)? Once addressed, re-request review from @dabla or re-mark the PR ready and it returns to the maintainer queue. Thank you.
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer β a real person β will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4662353518
Continuation/replacement for accidentally closed PR #65480 after branch history rewrite.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4572909853
@dabla @potiuk β branch rebased on latest main.
Static checks, docs, and unit test failures β
could you point me to the specific failing check
so I can fix it quickly? I want to get this
across the finish line. π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3289047110
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1040,3 +1109,46 @@ 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 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) via a thread executor.
+
+ Delegates to :meth:`SFTPHook.transfer` so that both the operator and the
+ trigger share identical transfer logic, in line with the DRY principle.
+ Using a thread executor keeps the async event loop unblocked while the
+ synchronous paramiko transfer runs in a worker thread.
+
+ :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).
+ """
+ import asyncio
+
+ loop = asyncio.get_running_loop()
+ sync_hook = SFTPHook(ssh_conn_id=self.sftp_conn_id)
+ await loop.run_in_executor(
+ None,
+ lambda: sync_hook.transfer(
+ operation=operation,
+ local_filepath=local_filepath,
+ remote_filepath=remote_filepath,
+ confirm=confirm,
+ create_intermediate_dirs=create_intermediate_dirs,
+ concurrency=concurrency,
+ prefetch=prefetch,
+ ),
+ )
Review Comment:
Just wondering, can't this just be written as:
```
sync_hook = SFTPHook(ssh_conn_id=self.sftp_conn_id)
asyncio.run(
sync_hook.transfer(
operation=operation,
local_filepath=local_filepath,
remote_filepath=remote_filepath,
confirm=confirm,
create_intermediate_dirs=create_intermediate_dirs,
concurrency=concurrency,
prefetch=prefetch,
),
)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3292615388
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1040,3 +1109,30 @@ 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 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:
+ import asyncio
Review Comment:
@dabla β both addressed:
1. SFTPOperatorTrigger renamed to SFTPTrigger (consistent with other providers)
2. SFTPHookAsync.transfer() now uses asgiref sync_to_async
3. asgiref>=3.5.2 added to provider dependencies
Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3279996174
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -139,3 +139,135 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperatorTrigger(BaseTrigger):
+ """
+ Trigger for SFTPOperator deferrable mode.
+
+ Fires when a file transfer (PUT, GET, or DELETE) completes
+ on the SFTP server, freeing the worker slot during the transfer.
+
+ :param ssh_conn_id: The SSH connection ID to use.
+ :param local_filepath: Local file path(s) to transfer.
+ :param remote_filepath: Remote file path(s) on the SFTP server.
+ :param operation: The SFTP operation - put, get, or delete.
+ :param confirm: Whether to confirm the file transfer.
+ :param create_intermediate_dirs: Whether to create intermediate dirs.
+ :param remote_host: Remote host to connect to (overrides connection).
+ :param concurrency: Number of threads for directory transfers.
+ :param prefetch: Whether to prefetch during file retrieval.
+ """
+
+ def __init__(
+ self,
+ ssh_conn_id: str | None = None,
+ local_filepath: str | list[str] | None = None,
+ remote_filepath: str | list[str] = "",
+ operation: str = "put",
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ remote_host: str | None = None,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ super().__init__()
+ self.ssh_conn_id = ssh_conn_id
+ self.local_filepath = local_filepath
+ self.remote_filepath = remote_filepath
+ self.operation = operation
+ self.confirm = confirm
+ self.create_intermediate_dirs = create_intermediate_dirs
+ self.remote_host = remote_host
+ self.concurrency = concurrency
+ self.prefetch = prefetch
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize the trigger for storage in the database."""
+ return (
+ "airflow.providers.sftp.triggers.sftp.SFTPOperatorTrigger",
+ {
+ "ssh_conn_id": self.ssh_conn_id,
+ "local_filepath": self.local_filepath,
+ "remote_filepath": self.remote_filepath,
+ "operation": self.operation,
+ "confirm": self.confirm,
+ "create_intermediate_dirs": self.create_intermediate_dirs,
+ "remote_host": self.remote_host,
+ "concurrency": self.concurrency,
+ "prefetch": self.prefetch,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Run the file transfer asynchronously and yield a TriggerEvent when done."""
+ try:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(
+ None,
+ self._do_transfer,
+ )
+ yield TriggerEvent(
+ {
+ "status": "success",
+ "local_filepath": self.local_filepath,
+ }
+ )
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+
+ def _do_transfer(self) -> None:
+ """Run the actual synchronous SFTP transfer in a thread executor."""
+ import os
+ from pathlib import Path
+
+ from airflow.providers.sftp.constants import SFTPOperation
+ from airflow.providers.sftp.hooks.sftp import SFTPHook
+
+ sftp_hook = SFTPHook(
Review Comment:
Also it doesn't make sense to use the `SFTPHook` which is synchronous in a trigger knowing we have a native `SFTPHookAsync` which has same capabilities of `SFTPHook`. We should use the `SFTPHookAsync` instead in trigger. Same remark regarding `_do_transfer`, both hooks should offer that functionality, so that trigger and operator only delegate the call to the hook. I would call the method `transfer` in `SFTPHook` and `SFTPHookAsync`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4301025395
@sunildataengineer This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Failing CI**: `Non-DB tests: core / Non-DB-core::3.10:Always`, `Low dep tests:core / All-core:LowestDeps:14:3.10:Always`, `CI image checks / Static checks`. Please investigate and fix.
**What to do next:**
- Address the items above.
- Make sure static checks pass locally: `prek run --from-ref main --stage pre-commit`.
- Mark the PR as "Ready for review" when you are done.
Converting a PR to draft is **not** a rejection β it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush β take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4474848668
@srchilukoori @potiuk @ashb β fixed the test_constants.py import path.
Was importing from: airflow.providers.sftp.operators.constants (wrong β doesn't exist in released wheel)
Now importing from: airflow.providers.sftp.constants (correct β matches actual module location)
This was causing all 8 CI failures across every Airflow version. Pushed and ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4513159762
@dabla β thank you for the detailed review and kind words!
Working on all 3 changes now. Will push shortly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3279996174
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -139,3 +139,135 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperatorTrigger(BaseTrigger):
+ """
+ Trigger for SFTPOperator deferrable mode.
+
+ Fires when a file transfer (PUT, GET, or DELETE) completes
+ on the SFTP server, freeing the worker slot during the transfer.
+
+ :param ssh_conn_id: The SSH connection ID to use.
+ :param local_filepath: Local file path(s) to transfer.
+ :param remote_filepath: Remote file path(s) on the SFTP server.
+ :param operation: The SFTP operation - put, get, or delete.
+ :param confirm: Whether to confirm the file transfer.
+ :param create_intermediate_dirs: Whether to create intermediate dirs.
+ :param remote_host: Remote host to connect to (overrides connection).
+ :param concurrency: Number of threads for directory transfers.
+ :param prefetch: Whether to prefetch during file retrieval.
+ """
+
+ def __init__(
+ self,
+ ssh_conn_id: str | None = None,
+ local_filepath: str | list[str] | None = None,
+ remote_filepath: str | list[str] = "",
+ operation: str = "put",
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ remote_host: str | None = None,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ super().__init__()
+ self.ssh_conn_id = ssh_conn_id
+ self.local_filepath = local_filepath
+ self.remote_filepath = remote_filepath
+ self.operation = operation
+ self.confirm = confirm
+ self.create_intermediate_dirs = create_intermediate_dirs
+ self.remote_host = remote_host
+ self.concurrency = concurrency
+ self.prefetch = prefetch
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize the trigger for storage in the database."""
+ return (
+ "airflow.providers.sftp.triggers.sftp.SFTPOperatorTrigger",
+ {
+ "ssh_conn_id": self.ssh_conn_id,
+ "local_filepath": self.local_filepath,
+ "remote_filepath": self.remote_filepath,
+ "operation": self.operation,
+ "confirm": self.confirm,
+ "create_intermediate_dirs": self.create_intermediate_dirs,
+ "remote_host": self.remote_host,
+ "concurrency": self.concurrency,
+ "prefetch": self.prefetch,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Run the file transfer asynchronously and yield a TriggerEvent when done."""
+ try:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(
+ None,
+ self._do_transfer,
+ )
+ yield TriggerEvent(
+ {
+ "status": "success",
+ "local_filepath": self.local_filepath,
+ }
+ )
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+
+ def _do_transfer(self) -> None:
+ """Run the actual synchronous SFTP transfer in a thread executor."""
+ import os
+ from pathlib import Path
+
+ from airflow.providers.sftp.constants import SFTPOperation
+ from airflow.providers.sftp.hooks.sftp import SFTPHook
+
+ sftp_hook = SFTPHook(
Review Comment:
Also it doesn't make sense to use the `SFTPHook` which is synchronous in a trigger knowing we have a native `SFTPHookAsync` which has same capabilities of `SFTPHook`. We should use the `SFTPHookAsync` instead in trigger. Same remark regarding `_do_transfer`, both hooks should offer that functionality, so that trigger and operator only delegate the call to the hook. I would call the method transfer in `SFTPHook` and `SFTPHookAsync`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4519511397
> @dabla β all 3 changes addressed:
>
> 1. serialize() now uses f"{self.**class**.**module**}.{self.**class**.**name**}"
> 2. Transfer logic moved to transfer() method on both SFTPHook and SFTPHookAsync
> 3. Trigger now uses SFTPHookAsync.transfer() β _do_transfer removed
>
> Ready for re-review π
Hello @sunildataengineer , if the addressed the remarks that I pointed out, you should resolve those conversations to make clear you resolved them, this makes it easier for us the reviewers, thx.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4468024967
@potiuk @ashb @gopidesupavan β all CI failures have been fixed
(missing test file, conf import, exception allowlist).
Branch is rebased on main and ready for review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "srchilukoori (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4301139573
@srchilukoori All review feedback has been addressed:
1. Extracted SFTPOperation into constants.py β circular import fixed
2. Added full type annotations to SFTPOperatorTrigger.__init__(),
serialize(), run(), _do_transfer()
3. Added remote_host, concurrency, prefetch to trigger __init__
and serialize()
4. Fixed directory transfer support in _do_transfer() β handles
isdir() check and calls retrieve_directory/store_directory correctly
5. Replaced asyncio.get_event_loop() with asyncio.get_running_loop()
6. Removed inner imports from test methods
7. Added SFTPOperatorTrigger tests: serialize() roundtrip,
run() success path, run() error path
All 6 tests passing locally. Ready for re-review.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4468001845
@srchilukoori β all CI failures fixed and pushed.
Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4544887698
@sunildataengineer You still have static checks, documentation and unit tests failing.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4506695963
Nice work @sunildataengineer, I had 3 remarks concerning this PR, so once those are resolved I think we're good as this will be a nice addition.
The reason why I insist on moving the logic of `_do_transfer` to a `transfer` method in both hooks is that when you need to process multiple files, the deferred operator won't help you. It's ideal when you only have to do one SFTP operation as then indeed you won't block the worker, but when multiple files (e.g. a LOT of files) need to be processed then it's best to use the async `PythonOperator` to allow multiplexing, which will indeed block one worker to process all files, but all sftp operations will be done in one event loop and you can use the [SFTPClientPool](https://github.com/apache/airflow/blob/main/providers/sftp/src/airflow/providers/sftp/pools/sftp.py#L46) which improves throughput, something you cannot do with multiple deferred operators. Of course this all depends on the use case, as I said, it's a good addition if you only need to do one specific SFTP operation.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on code in PR #65480:
URL: https://github.com/apache/airflow/pull/65480#discussion_r3292250624
##########
providers/sftp/src/airflow/providers/sftp/operators/sftp.py:
##########
@@ -108,8 +105,25 @@ def __init__(
self.remote_filepath = remote_filepath
self.concurrency = concurrency
self.prefetch = prefetch
+ self.deferrable = deferrable
def execute(self, context: Any) -> str | list[str] | None:
+ if self.deferrable:
+ self.defer(
+ trigger=SFTPOperatorTrigger(
+ ssh_conn_id=self.ssh_conn_id,
+ local_filepath=self.local_filepath,
+ remote_filepath=self.remote_filepath,
+ operation=self.operation,
+ confirm=self.confirm,
+ create_intermediate_dirs=self.create_intermediate_dirs,
+ remote_host=self.remote_host,
+ concurrency=self.concurrency,
+ prefetch=self.prefetch,
+ ),
+ method_name="execute_complete",
+ )
+
if self.local_filepath is None:
Review Comment:
@dabla β all addressed:
1. SFTPOperator.execute() now delegates to sftp_hook.transfer()
2. SFTPHookAsync.transfer() simplified to use asyncio.run()
3. Docstrings cleaned up
4. Conversations resolved
Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4524429754
@dabla β all addressed:
1. SFTPOperator.execute() now delegates to sftp_hook.transfer()
2. SFTPHookAsync.transfer() simplified to use asyncio.run()
3. Docstrings cleaned up
4. Conversations resolved
Ready for re-review π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4321689106
fixed all ruff linting issues
sorted import blocks in operator/sftp.py
removed duplicate sftp operator trigger imports from test methods
all 6 tests still passing locally
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer closed pull request #65480: Add deferrable mode to SFTPOperator
URL: https://github.com/apache/airflow/pull/65480
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4575006223
Approved your workflows. Just wait until CI will complete
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add deferrable mode to SFTPOperator [airflow]
Posted by "sunildataengineer (via GitHub)" <gi...@apache.org>.
sunildataengineer commented on PR #65480:
URL: https://github.com/apache/airflow/pull/65480#issuecomment-4642044639
@dabla @potiuk @srchilukoori @kaxil β fixed docs (asgiref added to requirements table) and added
newsfragment 65480.feature.rst. Pushing now. π
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org