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