s3.upload_fileobj() on arbitrary stream

[S-Boutot]

Hello,

I have seen this issue #2555 and this discussion is related but not totally the same problem.

I need to write a large set of data that I retrieve from an external API by chunks, this data is flexible and I don't know the size of it in advance.

This is why I use upload_fileobj() with the same chunk size in the multipart_chunksize parameter (5 MiB minimum as the doc says). I provide to the Fileobj parameter an io.BytesIO object with the read function implemented and I set seekable() to False as stated in the issue.

The read function works fine but once the API gave all the informations (egg. the length results in the chunk is lower that the chunk size) it should stop the reading and send the file to S3. But it doesn't and I am left with an infinite loop reading that last chunk.

After some retro engineering using a local file and the io.FileIO, It's sending the file to S3 once the length of the chunk received is lower than the chunk size.

If anyone as an idea of what I am missing, I would be grateful,

Stéphane.

[aBurmeseDev]

Hi @S-Boutot, thanks for reaching out and sorry to hear you’re having issues - in order for me to fully understand and reproduce the issue, it would be helpful to see the debug logs (by adding boto3.set_stream_logger('') to your code) and perhaps share your code snippet?

[S-Boutot]

Thanks for your help @aBurmeseDev,

Here is an exemple of my code where I replaced the api call by the function reader().

I added the boto3 logger in it but I don't see any errors. It will just not close the fileobj when it gets to b"the end".

from boto3.s3.transfer import TransferConfig
import io
import boto3
import logging

MB = 1024 * 1024
CHUNK_SIZE = 6 * MB

class Writer(io.BytesIO):
    def __init__(self, initial_bytes: bytes, data_reader=...) -> None:
        super().__init__(initial_bytes)
        self.data_reader = data_reader
        self.logger = logging.getLogger()

    def seekable(self):
        return False

    def read(self, __size: int = ...) -> bytes:
        """
        Everytime the read is called we get a new page from the API
        """
        data = self.data_reader()
        return data

class TestWriter():
    def __init__(self) -> None:
        self.upload_chunk_number = 0
        self.client = boto3.client(
            's3',
            aws_access_key_id=None,
            aws_secret_access_key=None,
        )

    boto3.set_stream_logger('')

    def reader(self, param=None):
        """
        Simulate a real return by the API
        """
        if self.upload_chunk_number == 3:
            return b"the end"
        else:
            self.upload_chunk_number += 1
            return b"a" * CHUNK_SIZE


    def main(self):
        data_stream = Writer(
            initial_bytes=b"a" * CHUNK_SIZE, data_reader=self.reader
        )

        transfer_config = TransferConfig(
            multipart_threshold=CHUNK_SIZE,
            multipart_chunksize=CHUNK_SIZE,
            num_download_attempts=5,
            max_io_queue=1,
            io_chunksize=CHUNK_SIZE,
            max_bandwidth=None,
            use_threads=False,
            max_concurrency=1,
        )

        self.client.upload_fileobj(
            Bucket="bucket-name", Key="key-name", Fileobj=data_stream, Config=transfer_config
        )


if __name__ == "__main__":
    app = TestWriter()
    # then call main function
    app.main()

[cvergnaud]

As you are running a non searchable data, you should return a b"" in your stop criteria code :
if self.upload_chunk_number == 3: return b""

[S-Boutot]

It's working indeed, thanks @cvergnaud