Skip to content

Instantly share code, notes, and snippets.

Last active February 2, 2019 21:57
  • Star 1 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save joyrexus/e892a7d4c380a47a91c713e61aa35f95 to your computer and use it in GitHub Desktop.
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)
self.head = iter([next(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()
seen += len(chunk)
if seen > size:
buf = None
seen = 0
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