Coverage for src/taipanstack/resilience/watchdogs/config_watcher.py: 100%
115 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Configuration watcher — detects file changes and hot-reloads config.
4Polls configuration files for modifications using SHA-256 hashes
5and validates new content via Pydantic before applying changes.
6"""
8import hashlib
9import json
10import logging
11import math
12from collections.abc import Callable, Sequence
13from pathlib import Path
15from pydantic import BaseModel, ValidationError
17from taipanstack.core.result import Err, Ok, Result
18from taipanstack.resilience.watchdogs._base import BaseWatcher
20logger = logging.getLogger("taipanstack.resilience.watchdogs.config")
22MAX_CONFIG_FILE_SIZE = 1048576
25def _hash_file(path: Path) -> Result[str, Exception]:
26 """Compute the SHA-256 hex digest of a file.
28 Args:
29 path: Path to the file.
31 Returns:
32 ``Ok(hex_digest)`` on success, ``Err`` on I/O failure.
34 """
35 try:
36 if path.stat().st_size > MAX_CONFIG_FILE_SIZE:
37 return Err(
38 ValueError(
39 f"File {path} exceeds max size ({MAX_CONFIG_FILE_SIZE} bytes)",
40 ),
41 )
42 data = path.read_bytes()
43 return Ok(hashlib.sha256(data).hexdigest())
44 except OSError as exc:
45 return Err(exc)
48def _is_valid_env_line(line: str) -> bool:
49 """Check if a line is a valid .env assignment."""
50 if not line or line.startswith("#"):
51 return False
52 return "=" in line
55def _parse_env_line(line: str) -> tuple[str, str]:
56 """Parse a valid .env assignment line into key and value."""
57 key, _, value = line.partition("=")
58 return key.strip(), value.strip().strip("\"'")
61def _parse_env(text: str) -> dict[str, object]:
62 """Parse a simple ``.env`` key=value file.
64 Lines starting with ``#`` or blank lines are skipped.
65 Surrounding quotes on values are stripped.
67 Args:
68 text: Raw text content of the ``.env`` file.
70 Returns:
71 Parsed key-value mapping.
73 """
74 result: dict[str, object] = {}
75 for line in text.splitlines():
76 stripped = line.strip()
77 if _is_valid_env_line(stripped):
78 key, value = _parse_env_line(stripped)
79 result[key] = value
80 return result
83def _parse_json(text: str) -> Result[dict[str, object], Exception]:
84 """Parse JSON text.
86 Args:
87 text: Raw JSON string.
89 Returns:
90 ``Ok(dict)`` on success, ``Err`` on parse failure.
92 """
93 try:
94 data = json.loads(text) # type: ignore[misc]
95 if not isinstance(data, dict): # type: ignore[misc]
96 return Err(TypeError(f"Expected JSON object, got {type(data).__name__}")) # type: ignore[misc]
97 return Ok(data)
98 except (json.JSONDecodeError, ValueError) as exc:
99 return Err(exc)
102def _read_file_content(path: Path) -> Result[str, Exception]:
103 """Read file content with size validation."""
104 try:
105 if path.stat().st_size > MAX_CONFIG_FILE_SIZE:
106 return Err(
107 ValueError(
108 f"File {path} exceeds max size ({MAX_CONFIG_FILE_SIZE} bytes)",
109 ),
110 )
111 return Ok(path.read_text(encoding="utf-8"))
112 except OSError as exc:
113 return Err(exc)
116def _parse_content_by_extension(
117 path: Path,
118 text: str,
119) -> Result[dict[str, object], Exception]:
120 """Parse file content based on extension."""
121 suffix = path.suffix.lower()
122 if suffix == ".json":
123 return _parse_json(text)
124 if suffix == ".env" or path.name == ".env":
125 return Ok(_parse_env(text))
127 return Err(ValueError(f"Unsupported config file extension: {suffix}"))
130def _load_file_data(path: Path) -> Result[dict[str, object], Exception]:
131 """Read and parse a configuration file based on its extension.
133 Supported extensions: ``.env``, ``.json``.
135 Args:
136 path: Path to the config file.
138 Returns:
139 ``Ok(dict)`` with parsed data, or ``Err`` on failure.
141 """
142 content_result = _read_file_content(path)
143 if isinstance(content_result, Err):
144 return content_result
146 return _parse_content_by_extension(path, content_result.ok_value)
149def validate_config(
150 data: dict[str, object],
151 model: type[BaseModel],
152) -> Result[BaseModel, Exception]:
153 """Validate a data dictionary against a Pydantic model.
155 Args:
156 data: Raw configuration data.
157 model: Pydantic model class to validate against.
159 Returns:
160 ``Ok(model_instance)`` on success, ``Err(ValidationError)``
161 on failure.
163 """
164 try:
165 return Ok(model.model_validate(data))
166 except ValidationError as exc:
167 return Err(exc)
170class ConfigWatcher(BaseWatcher):
171 """Background watcher that detects configuration file changes.
173 Polls file hashes at each interval. When a change is detected
174 the content is validated via the provided Pydantic model and,
175 if valid, the ``on_config_change`` callback is invoked.
177 Args:
178 config_paths: Files to watch.
179 config_model: Pydantic model for validation.
180 interval: Seconds between polls.
181 on_config_change: Callback receiving the validated model.
182 on_validation_error: Callback receiving the ``Exception``
183 when validation fails.
185 Example:
186 >>> watcher = ConfigWatcher(
187 ... config_paths=[Path(".env")],
188 ... config_model=MySettings,
189 ... on_config_change=lambda cfg: apply(cfg),
190 ... )
191 >>> await watcher.start()
193 """
195 def __init__(
196 self,
197 *,
198 config_paths: Sequence[Path],
199 config_model: type[BaseModel],
200 interval: float = 2.0,
201 on_config_change: Callable[[BaseModel], None] | None = None,
202 on_validation_error: Callable[[Exception], None] | None = None,
203 ) -> None:
204 """Initialize the config watcher.
206 Args:
207 config_paths: Files to watch.
208 config_model: Pydantic model for validation.
209 interval: Seconds between polls.
210 on_config_change: Callback for valid config changes.
211 on_validation_error: Callback for validation failures.
213 """
214 if (
215 not isinstance(interval, (int, float))
216 or not math.isfinite(interval)
217 or interval <= 0
218 ):
219 raise ValueError("interval must be a finite positive number")
220 super().__init__(interval=interval)
221 self._config_paths = list(config_paths)
222 self._config_model = config_model
223 self._on_config_change = on_config_change
224 self._on_validation_error = on_validation_error
225 self._file_hashes: dict[Path, str] = {}
227 def _process_hash_result(
228 self,
229 path: Path,
230 current_hash: str,
231 changed: list[Path],
232 ) -> None:
233 """Process successful hash result and update changed list."""
234 previous = self._file_hashes.get(path)
235 if previous is None:
236 # First time seeing this file — record hash
237 self._file_hashes[path] = current_hash
238 elif current_hash != previous:
239 self._file_hashes[path] = current_hash
240 changed.append(path)
242 def _detect_changes(self) -> Result[list[Path], Exception]:
243 """Detect which watched files have changed since last check.
245 Returns:
246 ``Ok(list[Path])`` of changed file paths.
248 """
249 changed: list[Path] = []
250 for path in self._config_paths:
251 hash_result = _hash_file(path)
252 if isinstance(hash_result, Ok):
253 self._process_hash_result(path, hash_result.ok_value, changed)
254 else:
255 logger.warning("Cannot hash %s: %s", path, hash_result.err_value)
256 return Ok(changed)
258 def _handle_validation_success(
259 self,
260 path: Path,
261 model: BaseModel,
262 ) -> Result[BaseModel, Exception]:
263 """Handle successful validation."""
264 logger.info(
265 "Config hot-reloaded from %s",
266 path,
267 )
268 if self._on_config_change is not None:
269 self._on_config_change(model)
270 return Ok(model)
272 def _handle_validation_failure(
273 self,
274 path: Path,
275 val_error: Exception,
276 ) -> Result[BaseModel, Exception]:
277 """Handle validation failure."""
278 logger.error(
279 "Config validation failed for %s: %s",
280 path,
281 val_error,
282 )
283 if self._on_validation_error is not None:
284 self._on_validation_error(val_error)
285 return Err(val_error)
287 def _validate_and_apply(self, path: Path) -> Result[BaseModel, Exception]:
288 """Load, validate, and apply configuration from a file.
290 Args:
291 path: Path to the changed config file.
293 Returns:
294 ``Ok(model)`` if valid, ``Err`` otherwise.
296 """
297 load_result = _load_file_data(path)
298 if isinstance(load_result, Err):
299 return Err(load_result.err_value)
300 validation = validate_config(load_result.ok_value, self._config_model)
301 if isinstance(validation, Ok):
302 return self._handle_validation_success(path, validation.ok_value)
303 return self._handle_validation_failure(path, validation.err_value)
305 async def _run(self) -> None:
306 """Execute a single config-check cycle."""
307 changes = self._detect_changes()
308 if isinstance(changes, Ok):
309 for path in changes.ok_value:
310 self._validate_and_apply(path)
311 else:
312 logger.error("Change detection failed: %s", changes.err_value)