Configuration & State Management

The Pattern

Problem: Your tool has user-configurable settings (host, port, auth mode) and runtime state (which sessions are active, device heartbeats) that must persist across restarts and handle concurrent reads/writes safely.

Approach: Separate config from state. Use a defaults-merge-overlay pattern with known-keys-only filtering for settings. Use atomic writes (write-to-tmp-then-os.replace) for state. Use asyncio locks for concurrent access. Follow XDG-conventional paths.

Pattern proven in production across multiple Python CLI tools and web services.

Key Design Decisions

1. Defaults-merge-overlay — never trust the file alone

The settings file might be from an older version (missing new keys) or a newer version (has keys we don't understand). The load_settings() pattern handles both:

def load_settings() -> dict:
    result = copy.deepcopy(DEFAULT_SETTINGS)    # start with ALL defaults
    try:
        text = SETTINGS_PATH.read_text()
        data = json.loads(text)
        for key in DEFAULT_SETTINGS:            # only copy KNOWN keys
            if key in data:
                result[key] = data[key]
    except (FileNotFoundError, json.JSONDecodeError):
        pass                                     # corrupt/missing = use defaults
    return result

The critical detail: iteration is over DEFAULT_SETTINGS keys, not over the file's keys. Unknown keys in the file are silently ignored. This prevents config drift when a user downgrades or when settings are synced between versions.

2. Known-keys-only filtering on write

The same principle applies when saving:

def save_settings(data: dict) -> None:
    merged = copy.deepcopy(DEFAULT_SETTINGS)
    for key in DEFAULT_SETTINGS:
        if key in data:
            merged[key] = data[key]
    SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
    SETTINGS_PATH.write_text(json.dumps(merged, indent=2) + "\n")

And on patch (partial update):

def patch_settings(patch: dict) -> dict:
    current = load_settings()
    for key in DEFAULT_SETTINGS:
        if key in patch:
            current[key] = patch[key]

3. Atomic writes — write-to-tmp-then-os.replace

State files can be read by other processes at any time. A naive write_text() can produce a half-written file if the process crashes mid-write.

The simple pattern:

def save_state(state: dict) -> None:
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    tmp = Path(str(STATE_PATH) + ".tmp")
    tmp.write_text(json.dumps(state, indent=2))
    os.replace(tmp, STATE_PATH)                 # atomic on POSIX

For extra safety (no predictable tmp path, proper cleanup on error), use tempfile.mkstemp:

def _write_instance(self, instance_id: str, data: dict) -> None:
    path = self._instance_path(instance_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    content = json.dumps(data, ensure_ascii=False, default=str)
    fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
    try:
        os.write(fd, content.encode("utf-8"))
        os.close(fd)
        Path(tmp_path).replace(path)
    except BaseException:
        with contextlib.suppress(OSError):
            os.close(fd)
        Path(tmp_path).unlink(missing_ok=True)
        raise

4. Asyncio lock for concurrent state access

When state is accessed from a poll loop and from API handlers simultaneously, a module-level asyncio lock serializes access:

state_lock: asyncio.Lock = asyncio.Lock()

async def read_state() -> dict:
    async with state_lock:
        return load_state()

async def write_state(state: dict) -> None:
    async with state_lock:
        save_state(state)

For threading contexts, use threading.Lock per instance with a defaultdict:

self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)

# Usage — every mutation acquires the per-instance lock:

…