[I] SFTPToGCSOperator Streaming unexpected BlobWritter [airflow]

["ignaciojcano (via GitHub)" <gi...@apache.org>]

ignaciojcano opened a new issue, #49801:
URL: https://github.com/apache/airflow/issues/49801

   ### Apache Airflow Provider(s)
   
   google
   
   ### Versions of Apache Airflow Providers
   
   apache-airflow-providers-google ==15.1.0
   apache-airflow-providers-sftp ==5.2.1
   
   
   
   ### Apache Airflow version
   
   2.10.2
   
   ### Operating System
   
   google cloud composer
   
   ### Deployment
   
   Google Cloud Composer
   
   ### Deployment details
   
   Composer Version 2.11.5
   Airflow Version 2.10.2
   
   ### What happened
   
   We are looking to use the SFTPToGCSOperator operator to stream files from a SFTP to a GCS bucket. Saw that the streaming for single file was added in https://github.com/apache/airflow/pull/48107 so tried to use it.
   
   The task was configured as shown below, but when executing got the error `Failed to execute job **** for task ***** (expected str, bytes or os.PathLike object, not BlobWriter; 33609)`
   
   ```
   fetch_task = SFTPToGCSOperator(
           task_id="task_id",
           sftp_conn_id="sftp_conn",
           gcp_conn_id="gcp_conn",
           source_path="/file.txt",
           destination_bucket="test_bucket",
           destination_path="path/to/file.txt",
           use_stream=True,
       )
   ```
   
   Looking at the code that  setups the stream, and the code in the sftp hook, it is basically expecting `write_stream` to be an instance of `BytesIO`, it is instead a `BlobWriter`, so it ends up calling `sftp.get` instead of `sftp.getfo)` 
   
   SFTP Hook retrieve_file: https://github.com/apache/airflow/blob/providers-sftp/5.1.2/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py#L277
   
   Storage "Blob": https://github.com/googleapis/python-storage/blob/main/google/cloud/storage/blob.py#L4098
   
   ### What you think should happen instead
   
   We should pass a `BytesIO` instance which represents the bucket object
   
   ### How to reproduce
   
   This is a pretty plan configuration of the operator, I cannot provide a bucket nor sftp server to test though.
   
   ```
   fetch_task = SFTPToGCSOperator(
           task_id="task_id",
           sftp_conn_id="sftp_conn",
           gcp_conn_id="gcp_conn",
           source_path="/file.txt",
           destination_bucket="test_bucket",
           destination_path="path/to/file.txt",
           use_stream=True,
       )
   ```
   
   ### Anything else
   
   _No response_
   
   ### Are you willing to submit PR?
   
   - [ ] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
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.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] SFTPToGCSOperator Streaming unexpected BlobWritter [airflow]

["potiuk (via GitHub)" <gi...@apache.org>]

potiuk closed issue #49801: SFTPToGCSOperator Streaming unexpected BlobWritter
URL: https://github.com/apache/airflow/issues/49801


-- 
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: [I] SFTPToGCSOperator Streaming unexpected BlobWritter [airflow]

["boring-cyborg[bot] (via GitHub)" <gi...@apache.org>]

boring-cyborg[bot] commented on issue #49801:
URL: https://github.com/apache/airflow/issues/49801#issuecomment-2831469205

   Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
   


-- 
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: [I] SFTPToGCSOperator Streaming unexpected BlobWritter [airflow]

["ignaciojcano (via GitHub)" <gi...@apache.org>]

ignaciojcano commented on issue #49801:
URL: https://github.com/apache/airflow/issues/49801#issuecomment-2835672016

   I looked at the underlying paramiko and it doesn't check if its a BytesIO implementation, so I think we could pass the BlobWritter directly, this means we would only use the `stfp_hook` to get the client and would ignore any future changes that might be done to `retrieve_file`
   
   ```
   # current
   with dest_blob.open("wb") as write_stream:
     sftp_hook.retrieve_file(source_path, write_stream, prefetch=self.sftp_prefetch)
   
   # proposed
   with dest_blob.open("wb") as write_stream:
     with sftp_hook.get_managed_conn() as conn:
       conn.getfo(source_path, write_stream, prefetch=self.sftp_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