Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 18 additions & 9 deletions bbot/core/config/files.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import sys
import yaml
from pathlib import Path
from omegaconf import OmegaConf

from .merge import deep_merge
from ...logger import log_to_stderr
from ...errors import ConfigLoadError

Expand All @@ -18,24 +19,32 @@ class BBOTConfigFiles:
def __init__(self, core):
self.core = core

def _get_config(self, filename, name="config"):
def _get_config(self, filename, name="config") -> dict:
filename = Path(filename).resolve()
if not filename.exists():
return {}
try:
conf = OmegaConf.load(str(filename))
with open(filename) as f:
conf = yaml.safe_load(f) or {}
if not isinstance(conf, dict):
raise ConfigLoadError(
f"Error parsing config at {filename}: expected a YAML mapping at the top level, "
f"got {type(conf).__name__}"
)
cli_silent = any(x in sys.argv for x in ("-s", "--silent"))
if __name__ == "__main__" and not cli_silent:
log_to_stderr(f"Loaded {name} from {filename}")
return conf
except ConfigLoadError:
raise
except Exception as e:
if filename.exists():
raise ConfigLoadError(f"Error parsing config at {filename}:\n\n{e}")
return OmegaConf.create()
raise ConfigLoadError(f"Error parsing config at {filename}:\n\n{e}")

def get_custom_config(self):
return OmegaConf.merge(
def get_custom_config(self) -> dict:
return deep_merge(
self._get_config(self.config_filename, name="config"),
self._get_config(self.secrets_filename, name="secrets"),
)

def get_default_config(self):
def get_default_config(self) -> dict:
return self._get_config(self.defaults_filename, name="defaults")
94 changes: 94 additions & 0 deletions bbot/core/config/merge.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
"""
Deep-merge helpers replacing omegaconf's merge semantics.

`deep_merge(a, b)` returns a new dict that is `a` with `b` merged in: nested
dicts are merged recursively, leaf values (and lists) from `b` replace those in
`a`. This matches `OmegaConf.merge(a, b)` for BBOT's preset layering use case.
"""

from __future__ import annotations

from copy import deepcopy
from typing import Any


def deep_merge(base: dict[str, Any] | None, *updates: dict[str, Any] | None) -> dict[str, Any]:
"""
Deep-merge one or more update dicts into a copy of `base`. Last wins on
leaf conflicts; lists are replaced wholesale (not concatenated).

The returned dict shares no mutable state with the inputs — nested dicts,
lists, and other mutable values are deep-copied as they're carried over.
"""
result: dict[str, Any] = deepcopy(base) if base else {}
for update in updates:
if not update:
continue
for k, v in update.items():
if k in result and isinstance(result[k], dict) and isinstance(v, dict):
result[k] = deep_merge(result[k], v)
else:
result[k] = deepcopy(v)
return result


def dotted_get(data: dict[str, Any], path: str, default: Any = None) -> Any:
"""
Look up a dotted path in a nested dict.

Note: keys containing literal `.` are not addressable (no escape syntax).

>>> dotted_get({"a": {"b": {"c": 1}}}, "a.b.c")
1
>>> dotted_get({"a": 1}, "a.b.c", default="x")
'x'
"""
cursor: Any = data
for part in path.split("."):
if not isinstance(cursor, dict) or part not in cursor:
return default
cursor = cursor[part]
return cursor


def dotted_set(data: dict[str, Any], path: str, value: Any) -> None:
"""
Set a dotted path in a nested dict, creating intermediate dicts as needed.

Non-dict intermediates are silently replaced. This is intentional —
callers (CLI parsing) feed the result through pydantic validation, which
surfaces any resulting type mismatch.

>>> d = {}
>>> dotted_set(d, "a.b.c", 1)
>>> d
{'a': {'b': {'c': 1}}}
"""
parts = path.split(".")
cursor = data
for part in parts[:-1]:
if part not in cursor or not isinstance(cursor[part], dict):
cursor[part] = {}
cursor = cursor[part]
cursor[parts[-1]] = value


def iter_dotted_paths(data: dict[str, Any], prefix: str = "") -> list[str]:
"""
Return every dotted leaf path in a nested dict. Empty dicts are treated
as leaves (so they round-trip through dotted_get/dotted_set).

>>> iter_dotted_paths({"a": 1, "b": {"c": 2}})
['a', 'b.c']
"""
paths: list[str] = []
for k, v in data.items():
path = f"{prefix}.{k}" if prefix else k
if isinstance(v, dict) and v:
paths.extend(iter_dotted_paths(v, path))
else:
paths.append(path)
return paths


__all__ = ["deep_merge", "dotted_get", "dotted_set", "iter_dotted_paths"]
Loading
Loading