File Locking for Concurrent CLI Runs in Python

A CLI that keeps state between runs — a sync marker, a download index, a queue, a counter — works perfectly until two copies run at once. Cron starts a job while the previous one is still going. A developer runs the same command in two terminals. A CI matrix fans out six jobs on one runner that share a cache. Each copy reads the state, does its work and writes the state back, and whichever writes last silently discards the other's changes. Nothing crashes and nothing is corrupted; data just goes missing. This guide shows how to put a lock around the dangerous section using the operating system's own file locking, how to decide between waiting and failing, and how to test that the lock actually works. It is part of the filesystem topic.

Prerequisites

• Python 3.10+ and the filelock package (uv add filelock), which works on Linux, macOS and Windows.
• A CLI with some shared state on disk — ideally already written with atomic writes.
• Familiarity with the idea of a critical section: the stretch of code that must not run in two processes at once.

The problem: lost updates

Atomic writes make every write complete. They say nothing about whether a write was based on current data. The classic failure is the read-modify-write race:

Both runs read {count: 5}, both compute 6, both write 6. The file is valid JSON, and one increment has vanished. Replace "count" with "list of files already uploaded" and the second run re-uploads everything; replace it with "last processed ID" and records are skipped. The only fix is to make the whole cycle — read, decide, write — exclusive.

Choosing a locking mechanism

There are three families of approach, and one important property separates them: what happens to the lock when the process holding it dies?

• Lock files created with O_EXCL. "If sync.lock exists, someone is running; otherwise create it and delete it at the end." Portable and simple, and broken by any crash, SIGKILL or power cut: the file stays, and every later run believes a phantom process holds the lock. You end up writing stale-lock detection with PIDs and timestamps, which has its own races.
• Advisory OS locks — fcntl.flock() on POSIX, msvcrt.locking() on Windows. The kernel tracks the lock against an open file descriptor and releases it automatically when the process exits for any reason. This is the property you want.
• The filelock package wraps the OS locks behind one cross-platform API with timeouts and a context manager, which makes it the pragmatic default. It also offers SoftFileLock, an O_EXCL-style lock for filesystems where OS locks do not work (some network filesystems).

The recipe

Lock the resource, not the program. Put the lock file next to the state it protects, and hold it for exactly the read-modify-write cycle:

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

import json
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Any

from filelock import FileLock, Timeout

from mytool.files import write_json_atomic   # from the atomic-writes guide


class Busy(Exception):
    """Another process holds the lock."""

    def __init__(self, lock_path: Path) -> None:
        super().__init__(f"another process is using {lock_path.parent}")
        self.lock_path = lock_path


@contextmanager
def locked_json(path: Path, *, timeout: float = 30.0) -> Iterator[dict[str, Any]]:
    """Load JSON under an exclusive lock; save atomically if the block succeeds."""
    path.parent.mkdir(parents=True, exist_ok=True)
    lock_path = path.with_name(path.name + ".lock")
    lock = FileLock(lock_path)
    try:
        lock.acquire(timeout=timeout)
    except Timeout:
        raise Busy(lock_path) from None
    try:
        data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
        yield data
        write_json_atomic(path, data)
    finally:
        lock.release()

A timeout of 0 means "try once and fail immediately", a positive number means "wait up to this long", and a negative one means "wait forever". The command layer maps those onto flags users understand:

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

import typer

from mytool.state import Busy, locked_json

app = typer.Typer()
STATE = Path.home() / ".local" / "state" / "mytool" / "sync.json"
EX_TEMPFAIL = 75


@app.callback()
def main() -> None:
    """Sync tool."""


@app.command()
def sync(
    wait: float = typer.Option(30.0, help="Seconds to wait for another run to finish."),
    no_wait: bool = typer.Option(False, "--no-wait", help="Fail at once if another run is active."),
) -> None:
    """Sync new items and record progress."""
    timeout = 0 if no_wait else wait
    try:
        with locked_json(STATE, timeout=timeout) as state:
            last = state.get("last_id", 0)
            new_items = list(range(last + 1, last + 4))   # stand-in for real work
            time.sleep(0.2)
            state["last_id"] = new_items[-1]
            state["runs"] = state.get("runs", 0) + 1
    except Busy as exc:
        typer.echo(f"error: {exc} (lock: {exc.lock_path})", err=True)
        raise typer.Exit(EX_TEMPFAIL)
    typer.echo(f"synced items {new_items[0]}..{new_items[-1]}", err=True)


if __name__ == "__main__":
    app()

Locking the resource versus locking the program

A "single instance" guard — one lock for the whole program, taken at startup — is sometimes what you want: a daemon-like watch command that must never run twice. More often it is too coarse. Two mytool sync --project a and --project b runs touch different state and should be free to run in parallel; a program-wide lock serialises them for no reason. Prefer one lock per resource you mutate, and add a program-wide lock only for commands that genuinely own the whole tool.

Where the lock file lives

Next to the data it protects, on the same filesystem, in a directory your tool owns — the state directory from storing app data with platformdirs is ideal. Do not delete lock files after releasing them; with OS locks, an existing unlocked file is harmless, and deleting it opens a race in which two processes lock two different inodes with the same name.

UX considerations

• Tell the user you are waiting. A command that silently blocks for thirty seconds looks hung. Print one line when the lock is not immediately available — try acquire(timeout=0) first, and only print before a blocking retry.
• Offer both behaviours. Interactive users usually prefer to wait a little; cron jobs and CI usually prefer to fail fast and try again on the next schedule. --wait SECONDS and --no-wait cover both.
• Exit with a "try again" code. 75 (EX_TEMPFAIL from sysexits.h) tells a scheduler that the failure is transient. It is more useful than a generic 1 — see choosing exit codes for CLI tools.
• Name the lock in the error. Printing the lock path lets an operator see which resource is contended and, if a network filesystem is misbehaving, find it.
• Keep critical sections short. Hold the lock only around the read-modify-write, not around a ten-minute download. If the work is long, read state, release, work, then re-lock and merge results — or accept that runs serialise, and say so in the help text.

Testing the behaviour

A lock test must use real concurrency — two processes, or at least two threads with separate lock objects — or it proves nothing. filelock locks are re-entrant per object within a process, so use separate FileLock instances, or better, separate processes:

# tests/test_state.py
import json
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path

import pytest
from filelock import FileLock

from mytool.state import Busy, locked_json


def bump(path_str: str) -> None:
    path = Path(path_str)
    for _ in range(50):
        with locked_json(path, timeout=10) as state:
            state["n"] = state.get("n", 0) + 1


def test_no_lost_updates_across_processes(tmp_path):
    path = tmp_path / "state.json"
    with ProcessPoolExecutor(max_workers=4) as pool:
        list(pool.map(bump, [str(path)] * 4))
    assert json.loads(path.read_text())["n"] == 200


def test_busy_when_locked_elsewhere(tmp_path):
    path = tmp_path / "state.json"
    other = FileLock(path.with_name("state.json.lock"))
    with other:
        with pytest.raises(Busy):
            with locked_json(path, timeout=0):
                pass


def test_error_in_block_writes_nothing(tmp_path):
    path = tmp_path / "state.json"
    with locked_json(path) as state:
        state["n"] = 1
    with pytest.raises(RuntimeError):
        with locked_json(path) as state:
            state["n"] = 99
            raise RuntimeError
    assert json.loads(path.read_text()) == {"n": 1}

Remove the lock from locked_json and the first test fails with a count well under 200 on almost every run — which is the demonstration that the lock is doing real work. Keep the worker function at module level so ProcessPoolExecutor can pickle it.

Conclusion

Atomic writes protect a file from crashes; locks protect it from your own tool running twice. Use OS-level locks — through filelock for portability — because they vanish with the process that held them, and scope each lock to the resource and the read-modify-write cycle that needs it. Let users choose between waiting and failing fast, exit with a retryable code when the lock is busy, and prove it all with a multi-process test.

Frequently asked questions

Related

• Up: Filesystem paths and atomic writes
• Writing files atomically in Python CLIs
• Storing app data with platformdirs
• Running a CLI on a schedule with cron and systemd
• Parallelising CLI work with thread pools