Writing Files Atomically in Python CLIs

A user runs mytool config set region eu-west-1, presses Ctrl+C a moment too early — or the laptop battery dies, or the disk fills — and the next run fails with TOMLDecodeError: Expected '=' after a key because config.toml is now empty. The code that wrote it looked fine: path.write_text(new_content). The problem is that writing a file in place is not one operation but several, and a failure between them leaves the file in a state that never should have existed. This guide builds a small, tested helper that makes every important write all-or-nothing, and shows where to use it in a CLI. It is the detailed companion to filesystem paths and atomic writes.

Prerequisites

• Python 3.10 or newer.
• A CLI that writes files it or its users care about: config, state, caches that are expensive to rebuild, generated outputs.
• pytest for the tests.

What goes wrong with an in-place write

Path.write_text() opens the file with mode "w", which truncates it to zero bytes immediately, then writes the new content, then closes it. Only after close() — and, for durability, after the operating system flushes its cache to disk — is the new content safely in place. Anything that stops the process between the truncate and the end leaves an empty or partial file: an exception, a KeyboardInterrupt, a SIGKILL from a timeout, the out-of-memory killer, a full disk, or a power cut.

The fix is a pattern databases and editors have used for decades. Write the new content to a different file, make sure it is completely on disk, then atomically rename that file over the original. On POSIX systems and on NTFS, a rename within one filesystem either happens entirely or not at all. Any reader — including your own tool on its next run — sees the complete old file or the complete new one.

The recipe

# src/mytool/files.py
from __future__ import annotations

import json
import os
import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import IO, Any


@contextmanager
def atomic_open(path: str | os.PathLike[str], mode: str = "w", *,
                encoding: str | None = "utf-8", newline: str | None = "",
                fsync: bool = True) -> Iterator[IO[Any]]:
    """Yield a file handle whose contents replace `path` only on success."""
    if mode not in ("w", "wb"):
        raise ValueError("atomic_open supports 'w' and 'wb'")
    target = Path(path).resolve()           # write through symlinks, not over them
    target.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.", suffix=".tmp")
    tmp = Path(tmp_name)
    try:
        if "b" in mode:
            fh = os.fdopen(fd, mode)
        else:
            fh = os.fdopen(fd, mode, encoding=encoding, newline=newline)
        with fh:
            yield fh
            fh.flush()
            if fsync:
                os.fsync(fh.fileno())
        try:
            os.chmod(tmp, target.stat().st_mode & 0o7777)   # keep existing permissions
        except FileNotFoundError:
            os.chmod(tmp, 0o666 & ~_umask())                # what open() would have used
        os.replace(tmp, target)
        if fsync and os.name == "posix":
            _fsync_dir(target.parent)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise


def write_text_atomic(path: str | os.PathLike[str], text: str, **kw: Any) -> None:
    with atomic_open(path, "w", **kw) as fh:
        fh.write(text)


def write_bytes_atomic(path: str | os.PathLike[str], data: bytes, **kw: Any) -> None:
    with atomic_open(path, "wb", encoding=None, newline=None, **kw) as fh:
        fh.write(data)


def write_json_atomic(path: str | os.PathLike[str], obj: Any) -> None:
    with atomic_open(path, "w") as fh:
        json.dump(obj, fh, indent=2, sort_keys=True, ensure_ascii=False)
        fh.write("\n")


def _umask() -> int:
    current = os.umask(0)
    os.umask(current)
    return current


def _fsync_dir(directory: Path) -> None:
    """Persist the rename itself (POSIX); a directory is a file of names."""
    dir_fd = os.open(directory, os.O_RDONLY)
    try:
        os.fsync(dir_fd)
    finally:
        os.close(dir_fd)

The context-manager form matters more than it looks. Serialisers such as json.dump, tomli_w.dump or csv.writer want a file handle, and building the whole output as a string first doubles peak memory for large files. With atomic_open, they write straight into the temporary file, and if the serialiser raises halfway — a value that is not JSON-serialisable, say — the exception propagates, the temporary file is deleted, and the original is untouched.

Why each step is there

• mkstemp(dir=target.parent) creates the temporary file in the same directory with a random name and O_EXCL, so it cannot collide with another process's temp file. The same directory guarantees the same filesystem; os.replace from /tmp to ~/.config would fail with EXDEV or silently become a non-atomic copy in a naive fallback.
• flush() then os.fsync() moves data from Python's buffer to the OS, then from the OS cache to the disk. Without fsync, a power loss shortly after the rename can leave a correctly named but empty file on some filesystems — the exact failure you were trying to prevent.
• Copying permissions preserves a 0600 credentials file as 0600. mkstemp always creates files as 0600, so without the chmod a world-readable config would quietly become private, and new files would not respect the user's umask.
• Path.resolve() first means that if the user's config is a symlink into their dotfiles repository, you replace the file the link points to rather than replacing the link with a regular file.
• os.replace overwrites an existing target on Windows too; os.rename raises FileExistsError there.
• Directory fsync makes the rename itself durable on POSIX. It is cheap and optional; include it for files that must survive a power cut.

Using it in commands

Replace in-place writes for anything that matters. A typical config command becomes:

# src/mytool/cli.py
import tomllib
from pathlib import Path

import tomli_w
import typer

from mytool.files import atomic_open

app = typer.Typer()
CONFIG = Path.home() / ".config" / "mytool" / "config.toml"


@app.callback()
def main() -> None:
    """mytool configuration."""


@app.command("set")
def set_value(key: str, value: str) -> None:
    """Set KEY to VALUE in the user config."""
    data = tomllib.loads(CONFIG.read_text(encoding="utf-8")) if CONFIG.exists() else {}
    data[key] = value
    with atomic_open(CONFIG, "wb", encoding=None, newline=None) as fh:
        tomli_w.dump(data, fh)
    typer.echo(f"{key} = {value!r}", err=True)


if __name__ == "__main__":
    app()

In a real tool, the config path would come from platformdirs rather than being hard-coded — see storing app data with platformdirs. And if two invocations might run set at once, the read-modify-write needs a lock as well; atomicity prevents torn files, not lost updates. File locking for concurrent CLI runs adds that.

UX considerations

• Output files the user asked for deserve the same care. mytool export -o report.csv that fails halfway should leave the previous report.csv intact, not a truncated one the user may not notice is incomplete.
• Don't leave dot-files behind. The temp files are hidden (leading dot) so they do not clutter a listing during the write, and the except branch removes them on failure. If you find stray .config.toml.*.tmp files, something bypassed the helper — usually a SIGKILL, which no code can intercept. Consider deleting stale ones older than a day on startup.
• Report the path you wrote. A single line on stderr ("wrote ~/.config/mytool/config.toml") confirms success and tells the user where to look.
• Keep writing to stdout streaming. Atomic writes are for files. When the user passes -o - for stdout, write directly; you cannot atomically replace a pipe.

Testing the behaviour

The key property is negative: when something fails mid-write, the old content survives and no temp file is left. Test it by failing inside the with block:

# tests/test_files.py
import json
import os
import stat
import sys

import pytest

from mytool.files import atomic_open, write_json_atomic, write_text_atomic


def test_replaces_content(tmp_path):
    p = tmp_path / "a.txt"
    p.write_text("old", encoding="utf-8")
    write_text_atomic(p, "new")
    assert p.read_text(encoding="utf-8") == "new"


def test_failure_keeps_old_file_and_cleans_up(tmp_path):
    p = tmp_path / "state.json"
    write_json_atomic(p, {"n": 1})
    with pytest.raises(TypeError):
        write_json_atomic(p, {"n": object()})   # not serialisable, fails mid-dump
    assert json.loads(p.read_text()) == {"n": 1}
    assert sorted(os.listdir(tmp_path)) == ["state.json"]


def test_interrupt_is_also_safe(tmp_path):
    p = tmp_path / "a.txt"
    p.write_text("keep me")
    with pytest.raises(KeyboardInterrupt):
        with atomic_open(p) as fh:
            fh.write("partial")
            raise KeyboardInterrupt
    assert p.read_text() == "keep me"


@pytest.mark.skipif(sys.platform == "win32", reason="POSIX permission bits")
def test_permissions_are_preserved(tmp_path):
    p = tmp_path / "secret.toml"
    p.write_text("token = 'x'")
    p.chmod(0o600)
    write_text_atomic(p, "token = 'y'")
    assert stat.S_IMODE(p.stat().st_mode) == 0o600


@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need privileges")
def test_symlink_target_is_updated(tmp_path):
    real = tmp_path / "dotfiles" / "config.toml"
    real.parent.mkdir()
    real.write_text("a = 1")
    link = tmp_path / "config.toml"
    link.symlink_to(real)
    write_text_atomic(link, "a = 2")
    assert link.is_symlink()
    assert real.read_text() == "a = 2"

You cannot easily simulate a power cut in a unit test, and you do not need to: the fsync calls are standard-library behaviour. Test your code's logic — cleanup, permissions, symlinks — and trust the operating system for the rest.

Conclusion

Every file your CLI writes that someone would miss should be written atomically: temporary file in the same directory, flush and fsync, preserve permissions, os.replace, and clean up on any exception. Wrapped in a context manager, it costs nothing at the call site — with atomic_open(path) as fh: instead of with open(path, "w") as fh: — and it turns a class of "my config vanished" bug reports into something that simply cannot happen short of hardware failure.

Frequently asked questions

Related

• Up: Filesystem paths and atomic writes
• File locking for concurrent CLI runs
• Safe temporary files and directories
• Storing app data with platformdirs
• Loading YAML configs safely in CLI apps