Feeding request.stream into tarfile.open(fileobj=...)

[bradbeattie]

I’m looking to use Starlette’s request.stream from a streaming upload and feed it straight into a tarfile.open similar to tarfile.open(fileobj=request.stream(), mode="r|").

But this doesn’t work as request.stream() is an async generator and tarfile.open is expecting a file-like object (e.g. io.BufferedReader). So this understandably errors out with

python3.10/tarfile.py", line 560, in __read
    buf = self.fileobj.read(self.bufsize)
AttributeError: 'async_generator' object has no attribute 'read'

Anyone have experiencing with converting an asynchronous generator to an io.BufferedReader? Or any other good way of supporting streaming uploads directly into tarfile.open?

[bradbeattie]

For anyone reading after-the-fact, a solution I found was to

• iterate over the request.stream(), putting the blocks into a queue
• have a separate thread that consumes the queue yielding bytes
• convert the generator into a buffered reader as per https://stackoverflow.com/a/51546783
• pass the buffered reader into tarfile.open(fileobj=...)

Not exactly elegant, but it gets the job done.

[Birnendampf]

There is a possibly more elegant way to do this that doesn't require a queue: Run the extraction in a separate thread and retrieve data from the stream using asyncio.run_coroutine_threadsafe(). Wrap this call with a dummy file object and pass it to tarfile.open.
This also makes it possible to very easily compute a checksum while downloading and extracting the file.
here is a sample implementation: https://gist.github.com/Birnendampf/93e03c4d503e0263499f9582c17719f5

[adriangb]

anyio has some pretty good facilities for this: https://anyio.readthedocs.io/en/stable/api.html#anyio.wrap_file

[bradbeattie]

Maybe I'm not seeing it, but if I want to effectively pipe the request.stream() (an async generator) into tarfile.open() (expects a synchronous fileobj), AnyIO doesn't have functionality I can use. Happy to be shown to be wrong though!

[adriangb]

I was thinking something like this:

import anyio

async def stream():
    yield b'hello!'

async def endpoint():
    with open('output.txt', 'wb') as f:
        f_async = anyio.wrap_file(f)
        async for chunk in stream():
            await f_async.write(chunk)

anyio.run(endpoint)

[bradbeattie]

Unfortunately not applicable as https://docs.python.org/3/library/tarfile.html#tarfile.open expects a synchronous fileobj. Look at lines like https://github.com/python/cpython/blob/3.12/Lib/tarfile.py#L2601. There's no awaiting in there!

[adriangb]

Right but you wrap the tarfile so you can write to it async. Awaiting doesn't need to happen within the tarfile.py code.

[bradbeattie]

Ah, to clarify I'm looking to take a request.stream and extract the contents to disk, not create a new tarfile.