config-state-patterns

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:
def update_instance(self, instance_id: str, **changes) -> InstanceStatus | None:
    with self._locks[instance_id]:
        data = self._read_instance(instance_id)
        ...

5. XDG-conventional paths

~/.config/my-tool/settings.json      # config
~/.config/my-tool/password            # secrets (0600)
~/.config/my-tool/secret              # signing key (0600)
~/.local/share/my-tool/state.json     # state
~/.my-tool/                           # all data (alternative)
~/.my-tool/token                      # auth token (0600)

A single env var override for the entire data root is useful:

def get_data_root() -> Path:
    env_dir = os.environ.get("MY_TOOL_DATA_DIR")
    root = Path(env_dir) if env_dir else Path.home() / ".my-tool"
    root.mkdir(parents=True, exist_ok=True)
    return root

Template / Starter Code

# settings.py
import copy, json, os
from pathlib import Path

SETTINGS_PATH = Path.home() / ".config" / "my-tool" / "settings.json"

DEFAULT_SETTINGS: dict = {
    "host": "127.0.0.1",
    "port": 8080,
    "log_level": "info",
}

def load_settings() -> dict:
    result = copy.deepcopy(DEFAULT_SETTINGS)
    try:
        data = json.loads(SETTINGS_PATH.read_text())
        for key in DEFAULT_SETTINGS:
            if key in data:
                result[key] = data[key]
    except (FileNotFoundError, json.JSONDecodeError):
        pass
    return result

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)
    tmp = Path(str(SETTINGS_PATH) + ".tmp")
    tmp.write_text(json.dumps(merged, indent=2) + "\n")
    os.replace(tmp, SETTINGS_PATH)

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

# state.py
import asyncio, contextlib, json, os, tempfile
from pathlib import Path

STATE_DIR = Path(os.environ.get("MY_TOOL_DATA_DIR",
                 str(Path.home() / ".local" / "share" / "my-tool")))
STATE_PATH = STATE_DIR / "state.json"
state_lock = asyncio.Lock()

def load_state() -> dict:
    try:
        return json.loads(STATE_PATH.read_text())
    except (FileNotFoundError, json.JSONDecodeError):
        return {"items": {}}

def save_state(state: dict) -> None:
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=STATE_DIR, suffix=".tmp")
    try:
        os.write(fd, json.dumps(state, indent=2).encode())
        os.close(fd)
        Path(tmp).replace(STATE_PATH)
    except BaseException:
        with contextlib.suppress(OSError):
            os.close(fd)
        Path(tmp).unlink(missing_ok=True)
        raise

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)

Gotchas & Lessons Learned

1. The choices-to-options merge regression. In one production system, PATCH /api/settings with nested objects would wipe secret keys because GET /api/settings redacts keys to "" for security. A naive merge overwrote real keys with empty strings. The fix preserves existing keys by identifier match, with a positional fallback for edits. This is a general hazard: any time you redact fields in a GET response, the PATCH handler must know not to treat redacted values as intentional changes.

2. defaultdict(threading.Lock) leaks memory. Per-instance locks are never pruned — one Lock (~100 bytes) per instance_id ever seen. This is acceptable for hundreds of instances but would need LRU eviction at scale.

3. copy.deepcopy(DEFAULT_SETTINGS) is critical. Without it, mutations to the returned dict would modify the module-level constant. This bug is invisible in single-call tests and only surfaces when settings are loaded twice in the same process.

4. File permissions for secrets: 0o600 after write, not on open(). Write the file first, then chmod(0o600). This avoids the race where another process reads the file between open() and write(). Also create the parent directory with 0o700.

5. JSON indent for human-editable files. Write indent=2 for config files. This lets users cat or vim their settings. State files that are only machine-read can skip indentation for smaller files and faster writes.