Add configurable confirm parameter to S3ToSFTPOperator

[harsh481]

Add configurable confirm parameter to S3ToSFTPOperator

Summary

This enhancement adds a confirm parameter to the S3ToSFTPOperator to make SFTP file transfer confirmation behavior explicit and configurable, providing consistency with SFTPOperator.

Changes

• Add confirm: bool = True parameter to S3ToSFTPOperator constructor.
• Pass confirm parameter explicitly to Paramiko's put() method.
• Update docstring with parameter documentation.
• Add comprehensive unit tests for both confirm=True and confirm=False scenarios.

Motivation

• One of the production jobs encountered an issue while dropping the file to the SFTP server using S3TOSFTPOperator. The server is configured to automatically move files to another designated directory when they are dropped at the root path.
• Since S3ToSFTPOperator works with Paramiko’s default setting confirm=True, it calls stat() to confirm a successful file transfer, which causes the pipeline to fail.

• This change allows users to:
  • Explicitly control confirmation behavior
  • Choose confirm=False for performance-critical scenarios
  • Have consistent API design with SFTPOperator.

Testing

• Added unit tests verifying both confirmation scenarios.
• All existing tests continue to pass.
• Code passes static analysis checks.

Backward Compatibility

This change is fully backward compatible. Existing code will continue to work exactly as before since the default value confirm=True matches Paramiko's existing default behavior.

[boring-cyborg]

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 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 Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment 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 for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• 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

[o-nikolas]

Barring passing tests, the changes look lgtm

[harsh481]

Hey @o-nikolas / @vincbeck
How do I merge these changes into the main branch?
Are there any additional steps I need to follow?

[vincbeck]

Hey @o-nikolas / @vincbeck How do I merge these changes into the main branch? Are there any additional steps I need to follow?

Hey @harsh481 , tests are failing. Can you please take a look and fix them? Examples of failures:

ERROR providers/amazon/tests/unit/amazon/aws/transfers/test_s3_to_sftp.py::TestS3ToSFTPOperator::test_s3_to_sftp_operation_confirm_true_default - airflow.exceptions.AirflowException: SSH operator error: exit status = 1
ERROR providers/amazon/tests/unit/amazon/aws/transfers/test_s3_to_sftp.py::TestS3ToSFTPOperator::test_s3_to_sftp_operation_confirm_false - airflow.exceptions.AirflowException: SSH operator error: exit status = 1

[harsh481]

Hey @o-nikolas / @vincbeck How do I merge these changes into the main branch? Are there any additional steps I need to follow?

Hey @harsh481 , tests are failing. Can you please take a look and fix them? Examples of failures:

ERROR providers/amazon/tests/unit/amazon/aws/transfers/test_s3_to_sftp.py::TestS3ToSFTPOperator::test_s3_to_sftp_operation_confirm_true_default - airflow.exceptions.AirflowException: SSH operator error: exit status = 1
ERROR providers/amazon/tests/unit/amazon/aws/transfers/test_s3_to_sftp.py::TestS3ToSFTPOperator::test_s3_to_sftp_operation_confirm_false - airflow.exceptions.AirflowException: SSH operator error: exit status = 1

Hey @vincbeck ,
Thank you for pointing out the issue in the test. I’ve fixed the SSH connection cleanup and made it consistent with test_s3_to_sftp_operation.
Here is the output of tests with latest commit changes:

• For test_s3_to_sftp_operation_confirm_false :

• For test_s3_to_sftp_operation_confirm_false :

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.