config-state-patterns

---
name: "config-state-patterns"
description: "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."
license: "MIT"
---

# 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:

```python
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:

```python
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):

```python
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:

```python
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`:

```python
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:

```python
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`:

```python
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:

```python
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

```python
# 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
```

```python
# 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.