test_ijson.py

# coding: utf-8

from __future__ import print_function

import resource
import gc
import time
import six

# set IJSON_BACKEND=yajl2_c or yajl2_cffi
import ijson


def memusage():
    """memory usage in MB. getrusage defaults to KB on Linux."""
    return str(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e3)


if six.PY2:
    from cStringIO import StringIO as BytesIO
else:
    from io import BytesIO

def check_memory(f):
    print("starting memory usage:", memusage(), 'MB')
    t_start = time.time()

    for i in range(400000):
        f()

    print("spent time: {:.2f}".format(time.time() - t_start))
    print("memory usage after ijson calls:", memusage(), 'MB')
    gc.collect()
    print("memory usage after garbage collection:", memusage(), 'MB')

if __name__ == "__main__":
    print("using backend", ijson.backend)
    print("using ijson version", ijson.__version__)

    data = BytesIO(b"""{"blah": {}}""")
    target = ijson.sendable_list()

    tests = [
        lambda: ijson.basic_parse(data),
        lambda: ijson.parse(data),
        lambda: ijson.items(data, "blah"),
        lambda: ijson.items(data, "blah"),

        lambda: ijson.basic_parse_coro(target),
        lambda: ijson.parse_coro(target),
        lambda: ijson.items_coro(target, "blah"),
        lambda: ijson.items_coro(target, "blah"),

        lambda: ijson.basic_parse_async(data),
        lambda: ijson.parse_async(data),
        lambda: ijson.items_async(data, "blah"),
        lambda: ijson.items_async(data, "blah"),
    ]

    for test in tests:
        check_memory(test)