Skip to content

Instantly share code, notes, and snippets.

@leplatrem leplatrem/async_pipe.py
Last active Aug 11, 2017

Embed
What would you like to do?
Download, decompress and process gzipped content
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.