rlsbl.lib.config_migrator

Generic config file migration engine.

Supports merging defaults into existing config files and running versioned migrations that mutate config values. All writes are atomic (tmp + rename).

Designed to be reusable across projects that need schema evolution for JSON config files.

Classes

ConfigMigrator

Generic config file migration engine.

Supports three merge strategies:

Usage: schema = { "files": [ {"path": "config.json", "defaults": {...}, "merge_strategy": "flat_dict"}, {"path": "theme.json", "defaults": {...}, "merge_strategy": "deep_recursive"}, {"path": "segments.json", "defaults": [...], "merge_strategy": "list_by_key", "match_field": "key"}, ], "schema_version_key": "_schema_version", "migrations": [ {"version": 1, "description": "...", "apply": some_callable}, ], } migrator = ConfigMigrator(schema) changes = migrator.run(Path("/path/to/config/dir"))

__init__

def __init__(self, schema: dict[str, Any]) -> None

Initialize with a schema describing files, defaults, and migrations.

Args: schema: dict with: - files: list of {path, defaults, merge_strategy, match_field?} - schema_version_key: str (default "_schema_version") - migrations: list of {version: int, description: str, apply: callable} where apply receives a dict of all loaded configs keyed by filename and mutates in place.

run

def run(self, base_dir: Path) -> dict[str, bool]

Run migrations on all files in base_dir.

Returns dict mapping filename -> whether it was written (changed).

deep_merge_missing

def deep_merge_missing(target: dict, defaults: dict) -> bool

Add missing keys from defaults recursively. Returns True if changed.

flat_merge_missing

def flat_merge_missing(target: dict, defaults: dict) -> bool

Add missing top-level keys. Returns True if changed.

list_merge_by_key

def list_merge_by_key(target_list: list[dict], defaults_list: list[dict], match_field: str) -> bool

For each default item, find match in target by match_field, add missing attrs.

Only enriches existing items in target_list. Does not add items that the user has removed (respects user deletions). Does not overwrite existing attrs.

_apply_migrations

def _apply_migrations(self, configs: dict[str, Any], schema_version_key: str) -> set[str]

Apply pending versioned migrations.

Returns set of filenames that were mutated by migrations. The schema version is stored in the first file in the schema's files list.

_save_json

def _save_json(path: Path, data: Any) -> None

Atomic write: tmp file + rename.

_load_json

def _load_json(path: Path) -> Any | None

Load JSON file, return None on missing/malformed.