Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
demo of a chunking json encoder
import json
import itertools
import StringIO
iterencode = json.JSONEncoder().iterencode
class SerializableList(list):
"""
Wraps an iterable to provide a JSON serializable generator.
"""
def __init__(self, iterable):
body = iter(iterable)
try:
self.head = iter([next(body)])
self.append(body)
except StopIteration:
self.head = []
def __iter__(self):
return itertools.chain(self.head, *self[:1])
def yield_in_chunks(iterable, size=4096):
buf = None
seen = 0
for chunk in iterable:
if buf is None:
buf = StringIO.StringIO()
buf.write(chunk)
seen += len(chunk)
if seen > size:
buf.seek(0)
yield buf.read()
buf = None
seen = 0
buf.seek(0)
yield buf.read()
def chunk_encode(iterable, size=4096):
return yield_in_chunks(iterencode(SerializableList(iterable)), size)
record = {"foo": {"bar": 'alpha'}}
results = (record for i in range(1000))
payload = "".join(chunk_encode(results))
assert type(payload) == str
assert len(json.loads(payload)) == 1000
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment