Add deferrable mode to SFTPOperator#68298
Conversation
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. - Add SFTPOperatorTrigger class in triggers/sftp.py - Add deferrable param to SFTPOperator.__init__() - Add defer() call in execute() when deferrable=True - Add execute_complete() callback method - Add unit tests for deferrable mode Closes apache#65475
Extract SFTPOperation into constants.py to fix circular import - Add full type annotations to SFTPOperatorTrigger - Add remote_host, concurrency, prefetch params to trigger - Fix directory transfer support in deferrable mode - Replace deprecated asyncio.get_event_loop() with get_running_loop() - Add remote_host, concurrency, prefetch to serialize() - Remove inner imports from test methods - Add SFTPOperatorTrigger tests: serialize roundtrip, run success, run error
sunildataengineer
requested review from
amoghrajesh,
ashb,
bugraoz93,
gopidesupavan,
jason810496,
jscheffl and
potiuk
as code owners
|
I’m opening a new PR as a continuation/replacement for accidentally closed PR #65480. While rebasing and cleaning up the branch history against upstream/main, I mistakenly performed a reset/rebase sequence that rewrote the branch history and caused GitHub to show “0 commits,” which made the original PR impossible to reopen properly. This was completely unintentional, and I sincerely apologize for the confusion and extra noise caused during review. Over the past couple of months, I worked extensively on this contribution — implementing deferrable support for Thankfully, the commits were still recoverable through Thank you again for all the reviews, feedback, patience, and guidance throughout this process. I truly appreciate the maintainers and reviewers taking another look at the contribution 🙏 |
|
@dabla @potiuk — the "200 changed files" in the GitHub Files tab The actual diff vs apache:main is only 11 files: providers/sftp/src/.../constants.py |
|
You could have simply renamed your old corrupt branch, and created a new branch with same name and force pushed it, that way the existing PR (with review and comments) would have been kept, which would make review for us easier. Now the context is unfortunately lost when creating a new PR, and we reviewers, have to check back with original PR. So please take that into consideration for the future, everyone makes mistakes me included, have encountered the same myself multiple times, that’s how I know the above trick is ideal in such situations. |
2 participants
Was generative AI tooling used to co-author this PR?
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.