Save JSON Files Safely with Atomic Writes in Python

A batch script finishes its work, opens progress.json for writing, and stops halfway through saving it. The next run finds an empty or incomplete file instead of the previous checkpoint.

For small JSON state files, write a complete replacement beside the destination and switch it into place only after the write succeeds. This tutorial implements that pattern using Python’s standard library and checks that a failed save preserves the previous contents.

1. Choose the save boundary

Use this approach for a small progress file, a generated configuration, or an experiment summary owned by one process. It fits alongside the dataset checksum workflow: a checksum identifies the inputs, while a progress file records how far a job has run.

The sequence is:

serialize JSON → write sibling temporary file → flush → fsync → close → replace

Python documents successful os.replace renames as atomic, with a POSIX requirement behind that guarantee. Keeping both paths in the same directory avoids a cross-filesystem move. Readers opening the destination on a suitable local filesystem see a complete old or new file; a reader that already opened the old file can continue reading it.

This example assumes a trusted directory, an ordinary destination path, and one writer. It is not a lock or a transaction across several files. Test your deployment’s filesystem semantics before using it on network storage.

2. Download and run the demo

Use Python 3.10 or newer. No packages are required. The example was tested with Python 3.14.7 on macOS.

Download atomic_json.py, save it in your working directory, and run:

python3 --version
python3 atomic_json.py --help
mkdir atomic-json-demo
python3 atomic_json.py atomic-json-demo/progress.json
python3 -m json.tool atomic-json-demo/progress.json

The demo intentionally replaces the output file, so use the new demo directory rather than an existing checkpoint. Its first two output lines are:

Rejected NaN; previous file unchanged
OK: completed_batches=2

The final file contains:

{
  "completed_batches": 2,
  "status": "ready"
}

The values are invented to demonstrate saving behavior. The demo does not run or resume a training job.

3. The complete implementation

"""Save a small JSON document by replacement. Python 3.10+, single writer."""
import argparse
import json
import os
from pathlib import Path
import tempfile


def atomic_write_json(path, data):
    """Replace a file in a trusted local directory; parent must already exist."""
    target = Path(path).absolute()
    # Serialize first: invalid data must not touch the destination.
    text = json.dumps(data, ensure_ascii=False, allow_nan=False, indent=2) + "\n"
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="\n",
            dir=target.parent, prefix=f".{target.name}.",
            suffix=".tmp", delete=False,
        ) as handle:
            temporary = Path(handle.name)
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        # Close before replacement, including on Windows.
        os.replace(temporary, target)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("output", type=Path, help="demo JSON path; replaced if present")
    args = parser.parse_args()
    old = {"completed_batches": 1, "status": "ready"}
    atomic_write_json(args.output, old)
    before = args.output.read_bytes()
    try:
        atomic_write_json(args.output, {"loss": float("nan")})
    except ValueError:
        assert args.output.read_bytes() == before
        print("Rejected NaN; previous file unchanged")
    new = {"completed_batches": 2, "status": "ready"}
    atomic_write_json(args.output, new)
    assert json.loads(args.output.read_text(encoding="utf-8")) == new
    print("OK: completed_batches=2")


if __name__ == "__main__":
    main()

Serialization happens before any file is created. json.dumps returns a string; allow_nan=False rejects non-finite floats instead of emitting the nonstandard NaN and Infinity values. This checks JSON encodability, not your application schema. Validate fields such as batch numbers before calling the function.

NamedTemporaryFile supplies a unique filename. Setting delete=False keeps it available after close, and dir=target.parent puts it beside the destination. The finally block removes the temporary file after an ordinary failure. Forced process termination can leave one behind.

The function holds the serialized text in memory. Keep it for small state documents; it is not a replacement for the large CSV streaming pattern.

4. Use it in your own script

Import the function from the downloaded module:

from pathlib import Path
from atomic_json import atomic_write_json

Path("run").mkdir(exist_ok=True)
atomic_write_json("run/progress.json", {
    "schema_version": 1,
    "completed_batches": 12,
    "input_manifest": "inputs-v1.json",
})

Save progress only after the corresponding output has been successfully committed. If a crash occurs between writing results and updating progress, the next run may repeat a batch. Make that operation safe to repeat, or use a storage design that commits results and progress together. Atomic replacement of this JSON file alone cannot coordinate them.

5. Verify preservation after failure

The demo tests a serialization failure. This separate check simulates a replacement failure after the temporary file has been written:

from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch
from atomic_json import atomic_write_json

with TemporaryDirectory() as directory:
    target = Path(directory) / "progress.json"
    atomic_write_json(target, {"completed_batches": 1})
    before = target.read_bytes()
    with patch("atomic_json.os.replace", side_effect=OSError("simulated failure")):
        try:
            atomic_write_json(target, {"completed_batches": 2})
        except OSError:
            pass
        else:
            raise AssertionError("Expected replacement to fail")
    assert target.read_bytes() == before
    assert list(Path(directory).iterdir()) == [target]
    print("PASS: old file preserved; temporary file removed")

This is an injected error test, not a power-loss test. flush followed by os.fsync sends the buffered file data through the operating system’s synchronization call. The helper does not sync the parent directory after replacement and makes no full power-loss durability guarantee. Linux’s fsync manual explains why the directory entry needs separate consideration.

6. Troubleshooting and limits

Temporary files use restrictive permissions on POSIX systems, and the replacement keeps the temporary file’s permissions. Do not use this helper unchanged for a shared configuration that depends on an existing owner, ACL, or mode. It also does not maintain a history: retain versioned checkpoints separately if you need rollback.

7. Takeaway

Serialize a small JSON document, write it beside the destination, and replace the destination only after the new file is complete. Test both valid saves and failures. Keep concurrency, recovery, and durability requirements explicit when this helper becomes part of a larger pipeline.