JSON编码很长的迭代器

Question

我正在编写一个Web服务,它返回包含很长列表的对象,这些列表以JSON编码.当然我们想要使用迭代器而不是Python列表,所以我们可以从数据库中流式传输对象; 遗憾的是,标准库(json.JSONEncoder)中的JSON编码器只接受要转换为JSON列表的列表和元组(虽然_iterencode_list看起来它实际上可以在任何迭代上工作).

文档字符串建议覆盖默认值以将对象转换为列表,但这意味着我们失去了流式传输的好处.以前,我们覆盖了一个私有方法,但是(正如预期的那样)在重构编码器时崩溃了.

以流方式将迭代器序列化为Python中的JSON列表的最佳方法是什么？

Answer 1

我确实需要这个。第一种方法是覆盖JSONEncoder.iterencode()方法。但是，这是行不通的，因为一旦迭代器不在顶层，某些_iterencode()函数的内部结构就会接管。

在研究了代码之后，我发现了一个非常棘手的解决方案，但是它可以工作。仅限于Python 3，但我确信python 2（只是其他魔术方法名称）可能具有相同的魔术效果：

import collections.abc
import json
import itertools
import sys
import resource
import time
starttime = time.time()
lasttime = None


def log_memory():
    if "linux" in sys.platform.lower():
        to_MB = 1024
    else:
        to_MB = 1024 * 1024
    print("Memory: %.1f MB, time since start: %.1f sec%s" % (
        resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / to_MB,
        time.time() - starttime,
        "; since last call: %.1f sec" % (time.time() - lasttime) if lasttime
        else "",
    ))
    globals()["lasttime"] = time.time()


class IterEncoder(json.JSONEncoder):
    """
    JSON Encoder that encodes iterators as well.
    Write directly to file to use minimal memory
    """
    class FakeListIterator(list):
        def __init__(self, iterable):
            self.iterable = iter(iterable)
            try:
                self.firstitem = next(self.iterable)
                self.truthy = True
            except StopIteration:
                self.truthy = False

        def __iter__(self):
            if not self.truthy:
                return iter([])
            return itertools.chain([self.firstitem], self.iterable)

        def __len__(self):
            raise NotImplementedError("Fakelist has no length")

        def __getitem__(self, i):
            raise NotImplementedError("Fakelist has no getitem")

        def __setitem__(self, i):
            raise NotImplementedError("Fakelist has no setitem")

        def __bool__(self):
            return self.truthy

    def default(self, o):
        if isinstance(o, collections.abc.Iterable):
            return type(self).FakeListIterator(o)
        return super().default(o)

print(json.dumps((i for i in range(10)), cls=IterEncoder))
print(json.dumps((i for i in range(0)), cls=IterEncoder))
print(json.dumps({"a": (i for i in range(10))}, cls=IterEncoder))
print(json.dumps({"a": (i for i in range(0))}, cls=IterEncoder))


log_memory()
print("dumping 10M numbers as incrementally")
with open("/dev/null", "wt") as fp:
    json.dump(range(10000000), fp, cls=IterEncoder)
log_memory()
print("dumping 10M numbers built in encoder")
with open("/dev/null", "wt") as fp:
    json.dump(list(range(10000000)), fp)
log_memory()

结果：

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[]
{"a": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]}
{"a": []}
Memory: 8.4 MB, time since start: 0.0 sec
dumping 10M numbers as incrementally
Memory: 9.0 MB, time since start: 8.6 sec; since last call: 8.6 sec
dumping 10M numbers built in encoder
Memory: 395.5 MB, time since start: 17.1 sec; since last call: 8.5 sec

很明显，IterEncoder不需要内存来存储10M int，同时保持相同的编码速度。

（hacky）的窍门是，_iterencode_list实际上并不需要任何列表中的东西。它只是想知道列表是否为空（__bool__），然后获取其迭代器。但是，只有在isinstance(x, (list, tuple))返回True 时，才进入此代码。因此，我将迭代器包装到一个列表子类中，然后禁用所有随机访问，将第一个元素放在前面，以便我知道它是否为空，并反馈迭代器。然后，default在使用迭代器的情况下，该方法返回此伪列表。