Counting bytes in a forwarded stream

[drtpotter]

Hi there,

In the documentation I see that I can open a connection from one source and pass the content to another stream as follows:

resp = await session.get('http://python.org')
await session.post('http://httpbin.org/post', data=resp.content)

In my implementation of this I will be forwarding streams of potentially very large files with unknown length. I need to be able to count the bytes that is transferred in resp.content without dumping resp.content to a local file. How can I do this?

Thankyou in advance!

[aio-libs-bot]

GitMate.io thinks the contributor most likely able to help you is @asvetlov.

Possibly related issues are #3440 (Streams reorganization), #1101 (Streaming to FormData), #2555 (Simplify stream classes), #3260 (Deprecate stream.unread_data), and #2680 (Request ignore X-Forwarded-Proto).

[asvetlov]

resp.content supports async iterator protocol.
You can wrap it into a custom class with counter:

class Wrapper:
    def __init__(self, content):
        self._content = content
        self._counter = 0
        self._iter = None

    def __aiter__(self):
         self._iter = self._content.__aiter__()
         return self._iter

    async def __anext__(self):
         chunk = await self._iter.__anext__()
         self._counter += len(chunk)
         return chunk

resp = await session.get('http://python.org')
wrapper = Wrapper(resp.content)
await session.post('http://httpbin.org/post', data=wrapper)
print(wrapper._counter)

[drtpotter]

Hi there,

Thanks for that, here is the full example but I can't seem to get anything other than zero out of wrapper._counter. Is wrapper deep-copied during session.post?

import aiohttp
import asyncio

class Wrapper:
    def __init__(self, content):
        self._content = content
        self._counter = 0
        self._iter = None

    def __aiter__(self):
         self._iter = self._content.__aiter__()
         return self  # return self, not self._iter

    async def __anext__(self):
         chunk = await self._iter.__anext__()
         self._counter += len(chunk)
         return chunk

async def main():
    session=aiohttp.ClientSession()
    resp = await session.get('http://python.org')
    wrapper = Wrapper(resp.content)
    await session.post('http://httpbin.org/post', data=wrapper)
    print(wrapper._counter)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

[asvetlov]

Sorry, I've made a mistake in __aiter__.
Updated your previous full example to make it work

[drtpotter]

Thanks for your help on this, much appreciated!