Python：逐个将 JSON / 字典对象写入文件

Question

我有一个大的 for 循环，在其中创建 JSON 对象，并希望能够在每次迭代中以流式方式将对象写入文件。我希望以后能够类似地使用该文件（一次读取一个对象）。

我的 JSON 对象包含换行符，不能将每个对象作为文件中的一行倾倒。

我该如何实现这个目标？

具体来说，考虑以下内容：

for _id in collection:
    dict_obj = build_dict(_id)  # build a dictionary object 
    with open('file.json', 'a') as f:
        stream_dump(dict_obj, f) 

stream_dump是我想要的函数。

请注意，我不想创建一个大列表并使用类似于json.dump(obj, file)这样的方法将整个列表转储。我希望能够在每次迭代中将对象附加到文件中。

谢谢。

- CentAu

---

如果我没有理解你的问题错误，似乎可以在每次写入对象后编写一个数据中不存在的分隔线，例如“-----”，并在读取时遇到该分隔线时创建一个新对象。 - alpert

啊，我明白了。那肯定可行。我以为可能还有其他的流处理解决方案。 - CentAu

Answer 1

您需要使用JSONEncoder的子类，并代理build_dict函数。

from __future__ import (absolute_import, division, print_function,)
#                        unicode_literals)

import collections
import json


mycollection = [1, 2, 3, 4]


def build_dict(_id):
    d = dict()
    d['my_' + str(_id)] = _id
    return d


class SeqProxy(collections.Sequence):
    def __init__(self, func, coll, *args, **kwargs):
        super(SeqProxy, *args, **kwargs)

        self.func = func
        self.coll = coll

    def __len__(self):
        return len(self.coll)

    def __getitem__(self, key):
        return self.func(self.coll[key])


class JsonEncoderProxy(json.JSONEncoder):
    def default(self, o):
        try:
            iterable = iter(o)
        except TypeError:
            pass
        else:
            return list(iterable)
        # Let the base class default method raise the TypeError
        return json.JSONEncoder.default(self, o)


jsonencoder = JsonEncoderProxy()
collproxy = SeqProxy(build_dict, mycollection)


for chunk in jsonencoder.iterencode(collproxy):
    print(chunk)

输出：

[
{
"my_1"
:
1
}
,
{
"my_2"
:
2
}
,
{
"my_3"
:
3
}
,
{
"my_4"
:
4
}
]

要逐块读取它，您需要使用JSONDecoder并将可调用的函数作为object_hook传递。当您调用JSONDecoder.decode(json_string)时，此钩子将被用于每个新解码的对象（您列表中的每个dict）。

- mementum

---

完美，谢谢。只是一个问题，SeqProxy是什么？ - CentAu

您的集合不会为每个项目返回一个“dict”（您正在每个项目上调用build_dict），而SeqProxy包装您的集合并在JSONEncoder请求序列化列表中的下一个项目时返回build_dict的结果。 - mementum

请纠正我如果我错了：这解决了两个问题：（a）需要代理来调用特定子集合上的自定义“build_dict”函数；（b）通过JSON模块提供的“iterencode”函数已经提供了逐块串行化的任务。- 我关注了（b），直到意识到它全部都是关于（a）。 - lenz

Answer 2

由于您自己生成文件，因此可以每行写出一个JSON对象：

for _id in collection:
    dict_obj = build_dict(_id)  # build a dictionary object 
    with open('file.json', 'a') as f:
        f.write(json.dumps(dict_obj))
        f.write('\n')

然后通过迭代行来读取它们：

with open('file.json', 'r') as f:
    for line in f:
        dict_obj = json.loads(line)

这不是一个很好的通用解决方案，但如果你既是生成者又是消费者，那么这是一个简单的解决方案。

- larsks

Answer 3

最简单的解决方案：

从您的JSON文档中删除所有空格字符：

import string

def remove_whitespaces(txt):
    """ We shall remove all whitespaces"""
    for chr in string.whitespace:
        txt = txt.replace(chr)

显然，您也可以使用json.dumps(json.loads(json_txt))（顺便说一下，这也验证了文本是否为有效的json）。

现在，您可以将文档每行写入文件。

第二种解决方案：

创建一个[AnyStr]Io流，在Io中编写有效文档（您的文档是对象或列表的一部分），然后将io写入文件（或将其上传到云端）。

- Fabrizio Ettore Messina

---

如果空格是内容的重要组成部分，会发生什么？ - mementum

好的观察！无论如何，json.dumps(json.loads(json_txt))在这种情况下都是完美的。 - Fabrizio Ettore Messina

为什么要删除所有空格？我不明白这与 OP 的联系。如果你想在单行上拥有完整的 JSON dump，请使用 json.dump(... indent=None)（实际上，它已经是默认设置了）。文本节点内的换行符会被转义。 - lenz