Download, decompress and process gzipped content

async_pipe.py

import asyncio
import io
import zlib
import sys

import asyncio
import aiohttp


async def download_big_file(loop):
    async with aiohttp.ClientSession(loop=loop) as session:
        url = "http://localhost:7777/firefox-2017-08-04T00-10Z.csv.gz"
        chunk_size = 1024

        gzip = zlib.decompressobj(zlib.MAX_WBITS | 16)

        async with session.get(url) as response:
            while "there are chunks to read":
                gzip_chunk = await response.content.read(chunk_size)
                if not gzip_chunk:
                    break
                csv_chunk = gzip.decompress(gzip_chunk)
                yield csv_chunk


async def csv2json(loop, async_input):
    leftover = ''
    async for csv_chunk in async_input:
        chunk = csv_chunk.decode("utf-8")
        lines = (leftover + chunk).split("\n")
        leftover = lines[-1]
        for line in lines[:-1]:
            fields = {i: f for i, f in enumerate(line.split(","))}
            yield fields


async def main(loop):
    async for l in csv2json(loop, download_big_file(loop)):
        print(l)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()