Coverage for src/taipanstack/resilience/watchdogs/config_watcher.py: 100%
109 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 15:01 +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
11from collections.abc import Callable, Sequence
12from pathlib import Path
14from pydantic import BaseModel, ValidationError
16from taipanstack.core.result import Err, Ok, Result
17from taipanstack.resilience.watchdogs._base import BaseWatcher
19logger = logging.getLogger("taipanstack.resilience.watchdogs.config")
21MAX_CONFIG_FILE_SIZE = 1048576
24def _hash_file(path: Path) -> Result[str, Exception]:
25 """Compute the SHA-256 hex digest of a file.
27 Args:
28 path: Path to the file.
30 Returns:
31 ``Ok(hex_digest)`` on success, ``Err`` on I/O failure.
33 """
34 try:
35 if path.stat().st_size > MAX_CONFIG_FILE_SIZE:
36 return Err(
37 ValueError(
38 f"File {path} exceeds max size ({MAX_CONFIG_FILE_SIZE} bytes)",
39 ),
40 )
41 data = path.read_bytes()
42 return Ok(hashlib.sha256(data).hexdigest())
43 except OSError as exc:
44 return Err(exc)
47def _parse_env(text: str) -> dict[str, object]:
48 """Parse a simple ``.env`` key=value file.
50 Lines starting with ``#`` or blank lines are skipped.
51 Surrounding quotes on values are stripped.
53 Args:
54 text: Raw text content of the ``.env`` file.
56 Returns:
57 Parsed key-value mapping.
59 """
60 result: dict[str, object] = {}
61 for line in text.splitlines():
62 stripped = line.strip()
63 if not stripped or stripped.startswith("#"):
64 continue
65 if "=" not in stripped:
66 continue
67 key, _, value = stripped.partition("=")
68 value = value.strip().strip("\"'")
69 result[key.strip()] = value
70 return result
73def _parse_json(text: str) -> Result[dict[str, object], Exception]:
74 """Parse JSON text.
76 Args:
77 text: Raw JSON string.
79 Returns:
80 ``Ok(dict)`` on success, ``Err`` on parse failure.
82 """
83 try:
84 data = json.loads(text)
85 if not isinstance(data, dict):
86 return Err(TypeError(f"Expected JSON object, got {type(data).__name__}"))
87 return Ok(data)
88 except (json.JSONDecodeError, ValueError) as exc:
89 return Err(exc)
92def _read_file_content(path: Path) -> Result[str, Exception]:
93 """Read file content with size validation."""
94 try:
95 if path.stat().st_size > MAX_CONFIG_FILE_SIZE:
96 return Err(
97 ValueError(
98 f"File {path} exceeds max size ({MAX_CONFIG_FILE_SIZE} bytes)",
99 ),
100 )
101 return Ok(path.read_text(encoding="utf-8"))
102 except OSError as exc:
103 return Err(exc)
106def _parse_content_by_extension(
107 path: Path,
108 text: str,
109) -> Result[dict[str, object], Exception]:
110 """Parse file content based on extension."""
111 suffix = path.suffix.lower()
112 if suffix == ".json":
113 return _parse_json(text)
114 if suffix == ".env" or path.name == ".env":
115 return Ok(_parse_env(text))
117 return Err(ValueError(f"Unsupported config file extension: {suffix}"))
120def _load_file_data(path: Path) -> Result[dict[str, object], Exception]:
121 """Read and parse a configuration file based on its extension.
123 Supported extensions: ``.env``, ``.json``.
125 Args:
126 path: Path to the config file.
128 Returns:
129 ``Ok(dict)`` with parsed data, or ``Err`` on failure.
131 """
132 content_result = _read_file_content(path)
133 if isinstance(content_result, Err):
134 return content_result
136 return _parse_content_by_extension(path, content_result.ok_value)
139def validate_config(
140 data: dict[str, object],
141 model: type[BaseModel],
142) -> Result[BaseModel, Exception]:
143 """Validate a data dictionary against a Pydantic model.
145 Args:
146 data: Raw configuration data.
147 model: Pydantic model class to validate against.
149 Returns:
150 ``Ok(model_instance)`` on success, ``Err(ValidationError)``
151 on failure.
153 """
154 try:
155 return Ok(model.model_validate(data))
156 except ValidationError as exc:
157 return Err(exc)
160class ConfigWatcher(BaseWatcher):
161 """Background watcher that detects configuration file changes.
163 Polls file hashes at each interval. When a change is detected
164 the content is validated via the provided Pydantic model and,
165 if valid, the ``on_config_change`` callback is invoked.
167 Args:
168 config_paths: Files to watch.
169 config_model: Pydantic model for validation.
170 interval: Seconds between polls.
171 on_config_change: Callback receiving the validated model.
172 on_validation_error: Callback receiving the ``Exception``
173 when validation fails.
175 Example:
176 >>> watcher = ConfigWatcher(
177 ... config_paths=[Path(".env")],
178 ... config_model=MySettings,
179 ... on_config_change=lambda cfg: apply(cfg),
180 ... )
181 >>> await watcher.start()
183 """
185 def __init__(
186 self,
187 *,
188 config_paths: Sequence[Path],
189 config_model: type[BaseModel],
190 interval: float = 2.0,
191 on_config_change: Callable[[BaseModel], None] | None = None,
192 on_validation_error: Callable[[Exception], None] | None = None,
193 ) -> None:
194 """Initialize the config watcher.
196 Args:
197 config_paths: Files to watch.
198 config_model: Pydantic model for validation.
199 interval: Seconds between polls.
200 on_config_change: Callback for valid config changes.
201 on_validation_error: Callback for validation failures.
203 """
204 super().__init__(interval=interval)
205 self._config_paths = list(config_paths)
206 self._config_model = config_model
207 self._on_config_change = on_config_change
208 self._on_validation_error = on_validation_error
209 self._file_hashes: dict[Path, str] = {}
211 def _process_hash_result(
212 self,
213 path: Path,
214 current_hash: str,
215 changed: list[Path],
216 ) -> None:
217 """Process successful hash result and update changed list."""
218 previous = self._file_hashes.get(path)
219 if previous is None:
220 # First time seeing this file — record hash
221 self._file_hashes[path] = current_hash
222 elif current_hash != previous:
223 self._file_hashes[path] = current_hash
224 changed.append(path)
226 def _detect_changes(self) -> Result[list[Path], Exception]:
227 """Detect which watched files have changed since last check.
229 Returns:
230 ``Ok(list[Path])`` of changed file paths.
232 """
233 changed: list[Path] = []
234 for path in self._config_paths:
235 hash_result = _hash_file(path)
236 if isinstance(hash_result, Ok):
237 self._process_hash_result(path, hash_result.ok_value, changed)
238 else:
239 logger.warning("Cannot hash %s: %s", path, hash_result.err_value)
240 return Ok(changed)
242 def _handle_validation_success(
243 self,
244 path: Path,
245 model: BaseModel,
246 ) -> Result[BaseModel, Exception]:
247 """Handle successful validation."""
248 logger.info(
249 "Config hot-reloaded from %s",
250 path,
251 )
252 if self._on_config_change is not None:
253 self._on_config_change(model)
254 return Ok(model)
256 def _handle_validation_failure(
257 self,
258 path: Path,
259 val_error: Exception,
260 ) -> Result[BaseModel, Exception]:
261 """Handle validation failure."""
262 logger.error(
263 "Config validation failed for %s: %s",
264 path,
265 val_error,
266 )
267 if self._on_validation_error is not None:
268 self._on_validation_error(val_error)
269 return Err(val_error)
271 def _validate_and_apply(self, path: Path) -> Result[BaseModel, Exception]:
272 """Load, validate, and apply configuration from a file.
274 Args:
275 path: Path to the changed config file.
277 Returns:
278 ``Ok(model)`` if valid, ``Err`` otherwise.
280 """
281 load_result = _load_file_data(path)
282 if isinstance(load_result, Err):
283 return Err(load_result.err_value)
284 validation = validate_config(load_result.ok_value, self._config_model)
285 if isinstance(validation, Ok):
286 return self._handle_validation_success(path, validation.ok_value)
287 return self._handle_validation_failure(path, validation.err_value)
289 async def _run(self) -> None:
290 """Execute a single config-check cycle."""
291 changes = self._detect_changes()
292 if isinstance(changes, Ok):
293 for path in changes.ok_value:
294 self._validate_and_apply(path)
295 else:
296 logger.error("Change detection failed: %s", changes.err_value)