python - JSON 编码的超长迭代器

Question

我正在编写一个 Web 服务，它返回包含非常长列表的对象，这些列表以 JSON 编码。当然，我们希望使用迭代器而不是 Python 列表，这样我们就可以从数据库中流式传输对象；不幸的是，标准库中的 JSON 编码器 (json.JSONEncoder) 只接受要转换为 JSON 列表的列表和元组(虽然 _iterencode_list 看起来它实际上适用于任何可迭代)。

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

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

Answer 1

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

在研究了代码之后，我发现了一个非常 hacky 的解决方案，但它确实有效。仅限 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 整数，同时保持相同的编码速度。

(hacky) 技巧是 _iterencode_list 实际上不需要任何列表内容。它只是想知道列表是否为空 (__bool__) 然后获取它的迭代器。然而，只有当 isinstance(x, (list, tuple)) 返回 True 时，它​​才会到达此代码。所以我将迭代器打包到一个列表子类中，然后禁用所有随机访问，在前面获取第一个元素以便我知道它是否为空，然后反馈迭代器。然后 default 方法在迭代器的情况下返回这个假列表。