Coverage for src/lilbee/core/config/resolve.py: 100%

1"""Each setting's effective value and the layer that supplies it: env > user > profile > built-in. 

2 

3Per-run CLI flags and ``apply_ephemeral_model_swap`` set cfg for one process and are not sources. 

4""" 

5 

6from __future__ import annotations 

7 

8import os 

9import tomllib 

10from collections.abc import Mapping 

11from dataclasses import dataclass 

12from pathlib import Path 

13from types import MappingProxyType 

14from typing import Any 

15 

16from pydantic.fields import FieldInfo 

17from pydantic_core import PydanticUndefined 

18 

19from .defaults import CONFIG_FILE_NAME, SKIP_TOML_ENV, env_var_name 

20from .enums import ProfileScope, SettingSource 

21from .load_warnings import warn_on_load 

22from .model import Config, default_refusal, hard_refusals, value_is_set 

23 

24PROFILE_TABLE = "profile" 

25PROFILE_NAME_KEY = "name" 

26PROFILE_VALUES_KEY = "values" 

27 

28# Built from data_root at construction; their pydantic default is an unresolved sentinel. 

29ROOT_DERIVED_FIELDS: frozenset[str] = frozenset( 

30 {"data_root", "documents_dir", "data_dir", "lancedb_dir", "models_dir"} 

31) 

32 

33 

34def _profile_scope(info: FieldInfo) -> ProfileScope | None: 

35 extra = info.json_schema_extra 

36 # pydantic types json_schema_extra as a dict or a callable; ConfigField always sets a dict 

37 scope = extra.get("profile") if isinstance(extra, dict) else None 

38 return None if scope is None else ProfileScope(str(scope)) 

39 

40 

41# The settings a profile may hold, each with the part of lilbee it tunes. 

42PROFILE_FIELDS: Mapping[str, ProfileScope] = MappingProxyType( 

43 { 

44 name: scope 

45 for name, info in Config.model_fields.items() 

46 if (scope := _profile_scope(info)) is not None 

47 } 

48) 

49 

50 

51@dataclass(frozen=True) 

52class ProfileTable: 

53 """The ``[profile]`` table of config.toml: the applied profile's name and recorded values.""" 

54 

55 name: str | None 

56 values: Mapping[str, Any] 

57 

58 

59@dataclass(frozen=True) 

60class SettingLayers: 

61 """The explicit layers read once: env vars, config.toml keys, and the ``[profile]`` values. 

62 

63 The user layer holds no value its setting refuses. ``refused_user`` names the config.toml 

64 keys left out for that reason. 

65 """ 

66 

67 env: Mapping[str, str] 

68 user: Mapping[str, Any] 

69 profile: Mapping[str, Any] 

70 refused_user: frozenset[str] = frozenset() 

71 

72 

73@dataclass(frozen=True) 

74class Resolved: 

75 """A setting's effective value and the layer it came from.""" 

76 

77 value: Any 

78 source: SettingSource 

79 

80 

81def _env_layer() -> dict[str, str]: 

82 """The set ``LILBEE_<FIELD>`` values, keyed by field name; a blank value counts as empty.""" 

83 layer: dict[str, str] = {} 

84 for name, info in Config.model_fields.items(): 

85 extra = info.json_schema_extra 

86 if isinstance(extra, dict) and extra.get("from_env") is False: 

87 continue # no environment string parses into this type, so no variable sets it 

88 raw = os.environ.get(env_var_name(name)) 

89 if raw is not None and value_is_set(name, raw.strip()): 

90 layer[name] = raw 

91 return layer 

92 

93 

94def _present(table: Mapping[str, Any]) -> dict[str, Any]: 

95 """Drop unset values: an empty string, except on a model role that can be off.""" 

96 return {key: value for key, value in table.items() if value_is_set(key, value)} 

97 

98 

99def _read_toml(path: Path) -> dict[str, Any]: 

100 """The parsed config.toml; empty when skipped, missing or unreadable. 

101 

102 The load and the overlay report an unreadable file; this read runs on every settings read. 

103 """ 

104 if os.environ.get(SKIP_TOML_ENV) == "1" or not path.exists(): 

105 return {} 

106 try: 

107 with path.open("rb") as f: 

108 return tomllib.load(f) 

109 except (ValueError, OSError): 

110 return {} 

111 

112 

113def toml_is_readable(path: Path) -> bool: 

114 """Whether the config.toml at *path* opens and parses.""" 

115 try: 

116 with path.open("rb") as f: 

117 tomllib.load(f) 

118 except (ValueError, OSError): 

119 return False 

120 return True 

121 

122 

123def _profile_table(data: Mapping[str, Any], path: Path) -> ProfileTable: 

124 """The ``[profile]`` table in *data*, keeping only the values a profile may hold.""" 

125 table = data.get(PROFILE_TABLE) 

126 # untyped TOML: a hand edit can put any type under [profile], its name or its values 

127 if not isinstance(table, dict): 

128 return ProfileTable(name=None, values={}) 

129 name = table.get(PROFILE_NAME_KEY) 

130 values = table.get(PROFILE_VALUES_KEY) 

131 kept: dict[str, Any] = {} 

132 for key, value in (values if isinstance(values, dict) else {}).items(): 

133 if key in PROFILE_FIELDS: 

134 kept[key] = value 

135 else: 

136 warn_on_load(f"Ignoring {key} in the profile values of {path}: profiles cannot set it") 

137 return ProfileTable(name=name if isinstance(name, str) else None, values=kept) 

138 

139 

140def read_profile_table(root: Path) -> ProfileTable: 

141 """The ``[profile]`` table of ``root/config.toml``.""" 

142 path = root / CONFIG_FILE_NAME 

143 return _profile_table(_read_toml(path), path) 

144 

145 

146def read_layers(root: Path) -> SettingLayers: 

147 """Read the env vars and ``root/config.toml`` once.""" 

148 return read_table_and_layers(root)[1] 

149 

150 

151def read_table_and_layers(root: Path) -> tuple[ProfileTable, SettingLayers]: 

152 """The ``[profile]`` table and the setting layers, from one read of ``root/config.toml``.""" 

153 path = root / CONFIG_FILE_NAME 

154 data = _read_toml(path) 

155 table = _profile_table(data, path) 

156 user = _present({key: value for key, value in data.items() if key != PROFILE_TABLE}) 

157 profile = _present(table.values) 

158 # A user value its setting refuses falls through to the next layer, so the source names the 

159 # layer that supplies the value. The load of config.toml reports it. A refused profile value 

160 # stays in its layer: a reset or a null write refuses to land on it. 

161 refused_user = hard_refusals(user) 

162 layers = SettingLayers( 

163 env=_env_layer(), 

164 user={k: v for k, v in user.items() if k not in refused_user}, 

165 profile=profile, 

166 refused_user=frozenset(refused_user), 

167 ) 

168 return table, layers 

169 

170 

171def builtin_value(key: str) -> Any: 

172 """The pydantic default for ``key``, or None when the field has none.""" 

173 info = Config.model_fields[key] 

174 if info.default_factory is not None: 

175 return info.default_factory() # type: ignore[call-arg] 

176 if info.default is PydanticUndefined: 

177 return None 

178 return info.default 

179 

180 

181def _is_derived(info: FieldInfo) -> bool: 

182 extra = info.json_schema_extra 

183 return isinstance(extra, dict) and bool(extra.get("derived", False)) 

184 

185 

186def resolve(key: str, layers: SettingLayers) -> Resolved: 

187 """The effective value of ``key`` and its source.""" 

188 for source, layer in ( 

189 (SettingSource.ENV, layers.env), 

190 (SettingSource.USER, layers.user), 

191 (SettingSource.PROFILE, layers.profile), 

192 ): 

193 if key in layer: 

194 return Resolved(layer[key], source) 

195 fallback = ( 

196 SettingSource.AUTO if _is_derived(Config.model_fields[key]) else SettingSource.BUILT_IN 

197 ) 

198 return Resolved(builtin_value(key), fallback) 

199 

200 

201def resolve_all(layers: SettingLayers) -> dict[str, Resolved]: 

202 """Resolve every Config field.""" 

203 return {key: resolve(key, layers) for key in Config.model_fields} 

204 

205 

206def _source_label(source: SettingSource, root: Path) -> str: 

207 """User-facing name for the file layer that supplied a value.""" 

208 if source is SettingSource.USER: 

209 return str(root / CONFIG_FILE_NAME) 

210 # SettingSource.PROFILE is the only other override layer a resolved value can carry. 

211 name = read_profile_table(root).name 

212 return f"the {name!r} profile" if name else "the applied profile" 

213 

214 

215def sanitize_soft_fields(resolved: dict[str, Resolved], root: Path) -> dict[str, Resolved]: 

216 """Replace a refused value of a setting that takes its default, and name its layer once. 

217 

218 A load of the same value reports it in the same words, so one process prints one line. 

219 """ 

220 sanitized = dict(resolved) 

221 labels: dict[SettingSource, str] = {} 

222 for key, entry in resolved.items(): 

223 if entry.source in (SettingSource.BUILT_IN, SettingSource.AUTO): 

224 continue 

225 refusal = default_refusal(key, entry.value) 

226 if refusal is None: 

227 continue 

228 if entry.source is SettingSource.ENV: 

229 warn_on_load(f"{env_var_name(key)} = {refusal}") 

230 else: 

231 if entry.source not in labels: 

232 labels[entry.source] = _source_label(entry.source, root) 

233 warn_on_load(f"{labels[entry.source]}: {key} = {refusal}") 

234 sanitized[key] = Resolved(builtin_value(key), entry.source) 

235 return sanitized