Python dict 被转储到 JSON 文件没有清除以前的文件内容

Question

在我的数据管道结束时，当我最终将 Python dict 推送到 JSON 文件以由 API 按需拉取时，我会将 dict 转储到文件中，如下所示：

json.dump(data, out_file)

在 99.9% 的情况下，这都能完美运行，并且最终用户可以以所需的格式访问数据。即：

out_file.json

{
    "good": {
        "JSON": "that", "I": "wanted", "to": ["push", ":)"]
    }, 
    "more_good": {
        "JSON": "that", "I": "wanted", "to": ["push", ":)"]
    }
}

但是，我的困难在于其他 0.1% 的推送......我一直注意到数据将被推送而不会从文件中完全删除以前的数据，我最终会遇到如下情况：

out_file.json

{
    "good": {
        "JSON": "that", "I": "wanted", "to": ["push", ":)"]
    }, 
    "more_good": {
        "JSON": "that", "I": "wanted", "to": ["push", ":)"]
    }
}ed", "to": ["push", ":)"]}}

截至目前，我想出了以下临时“解决方案”：

在推送 dict 之前，我会推送一个空字符串来清除文件：

json.dump('', out_file)
json.dump(data, out_file)

然后，在为最终用户获取文件内容时，我将检查以确保内容可用性，如下所示：

q = json.load(in_file)
while q == '': # also acts as an if
    q = json.load(in_file)
return q

我主要担心的是，在数据之前推送字符串只会降低边缘情况的可能性（即使是这样），并且我将继续看到这些相同的错误发生在未来 - 并增加了 end-不断向数据文件发送空白字符串会破坏用户数据的可访问性。

由于问题仅在 0.1% 的时间内发生，而且我不确定究竟是什么导致了极端情况，因此测试起来非常耗时，因此我无法确定我尝试的临时解决方案是否成功.无法测试边缘情况似乎本身就是一个错误 - 由于缺乏对导致错误的原因的理解。

Answer 1

您还没有展示out_file 是什么或如何打开它，但我预计问题在于两个线程/进程几乎同时尝试打开并写入文件。该文件在打开时被截断；所以如果顺序是 open1 - open2 - write1 - write2，你可能会得到类似的结果。有两种基本选择：

a) 如果另一个线程/进程正在做同样的事情，则使用某种锁定机制来发出错误信号：互斥锁、独占访问锁...然后您将不得不处理等待文件的线程之一不用了，还是放弃写吧。

b) 写入同一文件系统上的named temporary file，然后使用atomic replace。

我推荐第二种选择，既简单又安全。

Answer 2

我认为以上是正确的。您可以尝试的另一种解决方案是为文件创建一个唯一 ID，尽管我不确定这是否适用于您的用例并且它有点 hack 而不是解决根本原因。

import json
from uuid import uuid4


f = str(uuid4)
with open(data, 'w') as f:
    json.dump(data, f)

但显然，这仅在您不需要每次都调用该文件时才有效。

Answer 3

@Amardan 将问题诊断为由多个线程同时写入同一文件引起的，他一针见血。为了解决我的特定用例中的问题，我不得不稍微偏离他推荐的解决方案，甚至偶然地最终合并了@osint_alex 推荐的解决方案的元素。

不幸的是，在尝试使用 @Amardan 推荐的临时文件时。尝试时我会收到以下错误：

[Errno 18] Invalid cross-device link: '/tmp' -> '/app/data/out_file.json'

这不是什么大问题，因为解决方案真正在于能够以原子方式写入我的文件，而不是使用临时文件。因此，我所要做的就是在写入最终目的地之前创建自己的可访问文件作为数据的临时持有者。最终，我最终使用 UUID4 来命名这些临时文件，这样就不会同时写入两个文件（至少不会很快......）。最后，我实际上能够利用这个错误作为一个机会，将我所有的“json.dump”外包给一个函数，在那里我可以测试边缘情况并确保文件一次只写入一次。最后，新函数看起来像这样：

def update_content(content, dest):
    pth = f'/app/data/{uuid.uuid4()}.json'
    with open(pth, "w") as f:
        json.dump(content, f)
    try:
        with open(pth) as f:
            q = json.load(f)
        # NOTE: edge case testing here ...
        os.replace(pth, dest)
    except: # add exceptions as you see fit
        os.remove(pth)