tannnmayy opened a new pull request, #71557: URL: https://github.com/apache/airflow/pull/71557
Add optional deferrable mode to `SFTPOperator` so a single-file (or list of file pairs) get/put transfer can run on the Triggerer instead of occupying a worker slot for the whole transfer. closes: #65475 ## Why `SFTPOperator` is synchronous today. Large transfers hold a worker for the entire copy. `SFTPSensor` is already deferrable; this applies the same operator/trigger split to get/put. Three earlier attempts (#65480, #68520, #68298) did not land. #68298 was closed for a polluted diff and a security regression in `hooks/sftp.py`. This branch is a fresh rebase onto current `main` and contains only the feature. ## What this PR does - Add `deferrable` to `SFTPOperator`, defaulting to `[operators] default_deferrable` (`False` if unset) - Add `SFTPTransferTrigger` in the existing triggers module. `provider.yaml` already lists that module, so no extra registration is needed - `execute()` validates first, then `self.defer(...)` when `deferrable=True`. The existing synchronous transfer loop is not rewritten - `execute_complete()` resumes from the trigger event - Unit tests for serialize/run, deferral, guard rails, `execute_complete`, and the default sync path Unsupported in this first cut (raise `ValueError` before deferring): - `operation="delete"` - `concurrency > 1` - directory PUT (local path is a directory) - missing `ssh_conn_id` (a hook object cannot be serialized) ## Intentional limits - Local paths are read/written on the Triggerer host. Workers and Triggerer must share that filesystem - Directory GET is not pre-checked (that would require a worker-side SFTP call); `retrieve_file` fails at trigger time - `SFTPHookAsync.store_file` already creates missing remote parents, so PUT may create remote directories even when `create_intermediate_dirs=False` - `confirm` and `prefetch` are not on the async hook methods, so they are unused in the deferred path - `hooks/sftp.py` is untouched, including `_validate_within_directory` - No new `AirflowException` (allowlist stays at 2). Trigger/operator errors use `ValueError` - No `pyproject.toml`, generated docs, newsfragment, `constants.py`, or `uv.lock` changes ## Tests - Isolated checks of trigger serialize/run and operator defer/guard/`execute_complete` (this checkout cannot import Airflow on native Windows: `os.register_at_fork`, `_shared` gitlinks) - `ruff check` / `ruff format --check` on the four files - `git diff --check` CI on Linux should run: - `pytest providers/sftp/tests/unit/sftp/triggers/test_sftp.py` - `pytest providers/sftp/tests/unit/sftp/operators/test_sftp.py` Draft until CI is green. ##### Was generative AI tooling used to co-author this PR? - [x] Yes (Grok 4.6 / xAI) Generated-by: Grok 4.6 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
