Coverage for configzen/config.py: 32%
243 statements
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
1"""The base configuration model class [`BaseConfig`][configzen.config.BaseConfig]."""
3from __future__ import annotations 1abcdefgh
5from collections.abc import Callable, Iterable, Mapping 1abcdefgh
6from contextlib import suppress 1abcdefgh
7from contextvars import ContextVar 1abcdefgh
8from dataclasses import dataclass 1abcdefgh
9from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, cast 1abcdefgh
11from anyio.to_thread import run_sync 1abcdefgh
12from pydantic import BaseModel, PrivateAttr 1abcdefgh
13from pydantic._internal._config import config_keys as pydantic_config_keys 1abcdefgh
14from pydantic._internal._model_construction import ModelMetaclass 1abcdefgh
15from pydantic_settings import BaseSettings 1abcdefgh
16from pydantic_settings.main import SettingsConfigDict 1abcdefgh
18from configzen.context import isolated_context_coroutine, isolated_context_function 1abcdefgh
19from configzen.data import roundtrip_update_mapping 1abcdefgh
20from configzen.processor import ConfigProcessor, FileSystemAwareConfigProcessor 1abcdefgh
21from configzen.routes import ( 1abcdefgh
22 EMPTY_ROUTE,
23 GetAttr,
24 GetItem,
25 LinkedRoute,
26 Route,
27 RouteLike,
28 advance_linked_route,
29)
30from configzen.sources import ConfigSource, get_config_source 1abcdefgh
32if TYPE_CHECKING: 1abcdefgh
33 from collections.abc import Iterator
35 from pydantic import ConfigDict as BaseConfigDict
36 from typing_extensions import Self, Unpack
38 from configzen.data import Data
39 from configzen.routes import Step
42__all__ = ("BaseConfig", "ModelConfig") 1abcdefgh
45class ModelConfig(SettingsConfigDict, total=False): 1abcdefgh
46 """Meta-configuration for configzen models."""
48 config_source: str | ConfigSource[Any, Any] 1abcdefgh
49 rebuild_on_load: bool 1abcdefgh
50 processor_factory: Callable[..., ConfigProcessor] 1abcdefgh
53pydantic_config_keys |= set(ModelConfig.__annotations__) 1abcdefgh
54processing: ContextVar[ProcessingContext | None] = ContextVar( 1abcdefgh
55 "processing",
56 default=None,
57)
58owner_lookup: ContextVar[BaseConfig] = ContextVar("owner") 1abcdefgh
61class ProcessingContext(NamedTuple): 1abcdefgh
62 model_class: type[BaseConfig] 1abcdefgh
63 processor: ConfigProcessor 1abcdefgh
64 # We keep it mainly to make path resolution smarter.
65 trace: list[ConfigSource[Any, Any]] 1abcdefgh
68def _locate( 1abcdefgh
69 owner: object,
70 value: object,
71 route: Route,
72 subconfig: BaseConfig,
73) -> Iterator[Route]:
74 # Complex case: a subconfiguration in a subkey.
75 if value is owner:
76 return
77 attribute_access = False
78 if isinstance(value, BaseModel):
79 value = value.model_dump()
80 attribute_access = True
81 if isinstance(value, Mapping):
82 yield from _locate_in_mapping(
83 value,
84 subconfig,
85 route,
86 attribute_access=attribute_access,
87 ) # Complex case: a subconfiguration in an iterable.
88 elif isinstance(value, Iterable):
89 yield from _locate_in_iterable(value, subconfig, route)
92def _locate_in_mapping( 1abcdefgh
93 data: Mapping[Any, Any],
94 subconfig: BaseConfig,
95 base_route: Route = EMPTY_ROUTE,
96 *,
97 attribute_access: bool = False,
98) -> Iterator[Route]:
99 for key, value in data.items():
100 # Simple case: a subconfiguration at the current key.
101 route = base_route.enter(GetAttr(key) if attribute_access else GetItem(key))
102 if value is subconfig:
103 yield route
104 continue
105 # Complex case.
106 yield from _locate(
107 owner=data,
108 value=value,
109 route=route,
110 subconfig=subconfig,
111 )
114def _locate_in_iterable( 1abcdefgh
115 data: Iterable[object],
116 subconfig: BaseConfig,
117 base_route: Route = EMPTY_ROUTE,
118) -> Iterator[Route]:
119 for idx, value in enumerate(data):
120 # Simple case: a subconfiguration at the current index.
121 route = base_route.enter(GetItem(idx))
122 if value is subconfig:
123 yield route
124 continue
125 # Complex case.
126 yield from _locate(
127 owner=data,
128 value=value,
129 route=route,
130 subconfig=subconfig,
131 )
134class BaseConfigMetaclass(ModelMetaclass): 1abcdefgh
135 model_config: ModelConfig 1abcdefgh
137 if not TYPE_CHECKING: 137 ↛ exitline 137 didn't exit the body of class 'BaseConfigMetaclass', because the condition on line 137 was always true1abcdefgh
138 # Allow type-safe route declaration instead of using strings.
139 # Instead of writing conf.at("foo"), we can write conf.at(Conf.foo)
140 # to ensure full type safety backed by a membership check at runtime.
141 #
142 # Shoutout to Micael Jarniac for the suggestion.
144 def __getattr__(self, name: str) -> Any: 1abcdefgh
145 if name in self.model_fields: 1abcdefgh
146 return LinkedRoute(self, GetAttr(name)) 1abcdefgh
147 raise AttributeError(name) 1abcdefgh
150class BaseConfig(BaseSettings, metaclass=BaseConfigMetaclass): 1abcdefgh
151 """Base class for all configuration models."""
153 _config_source: ConfigSource[Any, Any] = PrivateAttr() 1abcdefgh
154 _config_data: Data = PrivateAttr(default_factory=dict) 1abcdefgh
155 _config_processor: ConfigProcessor = PrivateAttr() 1abcdefgh
156 _config_root: BaseConfig | None = PrivateAttr(default=None) 1abcdefgh
158 def __init__(self, **data: Any) -> None: 1abcdefgh
159 try: 1abcdefgh
160 owner = owner_lookup.get() 1abcdefgh
161 except LookupError: 1abcdefgh
162 owner = None 1abcdefgh
163 if processing.get(): 163 ↛ 164line 163 didn't jump to line 164, because the condition on line 163 was never true1abcdefgh
164 owner_lookup.set(self)
165 super().__init__(**data) 1abcdefgh
166 self._config_root = owner 1abcdefgh
168 # Mark the configzen's constructor as a non-custom constructor.
169 __init__.__pydantic_base_init__ = True # type: ignore[attr-defined] 1abcdefgh
171 @property 1abcdefgh
172 def config_root(self) -> BaseConfig: 1abcdefgh
173 """Return the root configuration that was used to load the entire data."""
174 return self._config_root or self
176 @property 1abcdefgh
177 def config_source(self) -> ConfigSource[Any, Any] | None: 1abcdefgh
178 """Return the configuration source that was used to load the configuration."""
179 if self._config_root is None:
180 # Since _config_source is a private attribute
181 # without a default value, we need to use getattr
182 # to avoid an AttributeError in case this attribute
183 # was not set (which may happen when the configuration
184 # is instantiated manually).
185 return getattr(self, "_config_source", None)
186 return self._config_root.config_source
188 @property 1abcdefgh
189 def config_data(self) -> Data: 1abcdefgh
190 """Return the configuration that was loaded from the configuration source."""
191 if self._config_root is None:
192 return self._config_data
193 return self._config_root.config_data
195 @property 1abcdefgh
196 def config_processor(self) -> ConfigProcessor: 1abcdefgh
197 """
198 Current configuration processor.
200 Processor stores the initial data used when loading the configuration,
201 resolves macros etc.
202 """
203 if self._config_root is None:
204 if not hasattr(self, "_config_processor"):
205 return FileSystemAwareConfigProcessor(self.config_dump())
206 return self._config_processor
207 return self._config_root.config_processor
209 def config_find_routes( 1abcdefgh
210 self,
211 subconfig: BaseConfig,
212 ) -> set[Route]:
213 """
214 Locate all occurrences of a subconfiguration in the current configuration.
216 Return a set of routes to the located subconfiguration.
217 """
218 if not isinstance(subconfig, BaseConfig):
219 msg = f"Expected a BaseConfig subclass instance, got {type(subconfig)!r}"
220 raise TypeError(msg)
221 return set(
222 _locate_in_mapping(vars(self), subconfig, attribute_access=True),
223 )
225 def config_find_route(self, subconfig: BaseConfig) -> Route: 1abcdefgh
226 """Locate exactly one (closest) route to the given subconfiguration."""
227 all_routes = self.config_find_routes(subconfig)
228 if not all_routes:
229 msg = f"Unable to locate subconfiguration {subconfig}"
230 raise LookupError(msg)
231 return next(iter(all_routes))
233 @classmethod 1abcdefgh
234 def _validate_config_source( 1abcdefg
235 cls,
236 source: object | None = None,
237 ) -> ConfigSource[Any, Any]:
238 if source is None:
239 source = cls.model_config.get("config_source")
240 if source is None:
241 msg = f"No config source provided when loading {cls.__name__}"
242 raise ValueError(msg)
243 if not isinstance(source, ConfigSource):
244 source = get_config_source(source)
245 if source is None:
246 msg = (
247 f"Could not create a config source from {source!r} "
248 f"of type {type(source)!r}"
249 )
250 raise ValueError(msg)
251 return source
253 @classmethod 1abcdefgh
254 def _validate_processor_factory( 1abcdefg
255 cls,
256 processor_factory: Callable[..., ConfigProcessor] | None = None,
257 ) -> Callable[..., ConfigProcessor]:
258 return (
259 processor_factory
260 or cast(
261 "Callable[..., ConfigProcessor] | None",
262 cls.model_config.get("config_processor_factory"),
263 )
264 or FileSystemAwareConfigProcessor
265 )
267 @classmethod 1abcdefgh
268 def _try_rebuild_model(cls) -> None: 1abcdefgh
269 # Possible scenarios:
270 # (sync) Frame 1: <class>.config_load()
271 # (sync) Frame 2: isolated_context_function.<locals>.copy()
272 # (sync) Frame 3: run_isolated()
273 # (sync) Frame 4: <class>.config_load()
274 # (sync) Frame 5: <class>.model_rebuild()
275 #
276 # (async) Frame 1: <class>.config_load_async()
277 # (async) Frame 2: isolated_context_function.<locals>.copy()
278 # (async) Frame 3: run_isolated()
279 # (async) Frame 4: <class>.config_load()
280 # (async) Frame 5: <class>.model_rebuild()
281 if cls.model_config["rebuild_on_load"]:
282 with suppress(Exception):
283 cls.model_rebuild(_parent_namespace_depth=5)
285 @classmethod 1abcdefgh
286 @isolated_context_function 1abcdefgh
287 def config_load( 1abcdefg
288 cls,
289 source: object | None = None,
290 *,
291 processor_factory: Callable[..., ConfigProcessor] | None = None,
292 ) -> Self:
293 """
294 Load this configuration from a given source.
296 Parameters
297 ----------
298 source
299 Where to load the configuration from. The argument passed is forwarded
300 to `confizen.sources.get_config_source()` which will resolve
301 the intended configuration source: for example, "abc.ini" will be resolved
302 to a TOML text file source. Keep in mind, however, that for binary formats
303 such as non-XML Plist you must specify its format type to binary, so in
304 that case just create `BinaryFileConfigSource("plist_file.plist")`.
305 processor_factory
306 The state factory to use to parse the newly loaded configuration data.
308 Returns
309 -------
310 self
312 """
313 cls._try_rebuild_model()
315 # Validate the source we load our configuration from.
316 config_source = cls._validate_config_source(source)
318 # Validate the processor we use to parse the loaded configuration data.
319 make_processor = cls._validate_processor_factory(processor_factory)
321 # Load the configuration data from the sanitized source.
322 # Keep in mind the loaded data object keeps all the additional
323 # metadata that we want to keep.
324 # Then we pass it to the processor factory to process the configuration data
325 # into a bare dictionary that does not hold anything else
326 # than the configuration data, by using `processor.get_processed_data()`.
327 processor = make_processor(config_source.load())
329 processing_context = ProcessingContext(cls, processor, trace=[config_source])
330 processing.set(processing_context)
331 # ruff: noqa: FBT003
332 try:
333 # Processing will execute any commands that are present
334 # in the configuration data and return the final configuration
335 # data that we will use to construct an instance of the configuration model.
336 # During this process, we lose all the additional metadata that we
337 # want to keep in the configuration data.
338 # They will be added back to the exported data when the configuration
339 # is saved (`processor.revert_processor_changes()`).
340 self = cls(**processor.get_processed_data())
341 finally:
342 processing.set(None)
344 # Quick setup and we're done.
345 self._config_source = config_source
346 self._config_processor = processor
347 return self
349 @classmethod 1abcdefgh
350 @isolated_context_coroutine 1abcdefgh
351 async def config_load_async( 1abcdefg
352 cls,
353 source: object | None = None,
354 *,
355 processor_factory: Callable[..., ConfigProcessor] | None = None,
356 ) -> Self:
357 """
358 Do the same as `config_load`, but asynchronously (no I/O blocking).
360 Parameters
361 ----------
362 source
363 Where to load the configuration from. The argument passed is forwarded
364 to `confizen.sources.get_config_source()` which will resolve
365 the intended configuration source: for example, "abc.ini" will be resolved
366 to a TOML text file source. Keep in mind, however, that for binary formats
367 such as non-XML Plist you must specify its format type to binary, so in
368 that case just create `BinaryFileConfig"plist_file.plist")`.
369 processor_factory
370 The state factory to use to parse the newly loaded configuration data.
372 Returns
373 -------
374 self
376 """
377 cls._try_rebuild_model()
379 # Intentionally not using `run_sync(config_load)` here.
380 # We want to keep make the set up instructions blocking to avoid running
381 # into mutexes.
383 config_source = cls._validate_config_source(source)
384 make_processor = cls._validate_processor_factory(processor_factory)
385 processor = make_processor(await config_source.load_async())
387 processing_context = ProcessingContext(cls, processor, trace=[config_source])
388 processing.set(processing_context)
389 try:
390 self = cls(**await run_sync(processor.get_processed_data))
391 finally:
392 processing.set(None)
394 self._config_processor = processor
395 self._config_source = config_source
396 return self
398 def config_reload(self) -> Self: 1abcdefgh
399 """Reload the configuration from the same source."""
400 source = self.config_source
402 if source is None:
403 msg = "Cannot reload a manually instantiated configuration"
404 raise RuntimeError(msg)
406 root = self.config_root
408 # Create a new processor with the same options as the current one.
409 processor = root.config_processor.create_processor(source.load())
411 # Construct a new configuration instance.
412 # Respect __class__ attribute in case root might be a proxy (from proxyvars).
413 new_root = root.__class__(**processor.get_processed_data())
415 # Copy values from the freshly loaded configuration into our instance.
416 if root is self:
417 new_data = new_root.config_dump()
418 else:
419 route_to_self = root.config_find_route(self)
420 new_data = cast("Self", route_to_self.get(new_root)).config_dump()
422 for key, value in new_data.items():
423 setattr(self, key, value)
425 return self
427 async def config_reload_async(self) -> Self: 1abcdefgh
428 """Do the same as `config_reload` asynchronously (no I/O blocking)."""
429 source = self.config_source
431 if source is None:
432 msg = "Cannot reload a manually instantiated configuration"
433 raise RuntimeError(msg)
435 root = self.config_root
437 # Create a new state processor the same options as the current one.
438 processor = root.config_processor.create_processor(source.load())
440 # Construct a new configuration instance.
441 new_root = root.__class__(**await run_sync(processor.get_processed_data))
443 # Copy values from the freshly loaded configuration into our instance.
444 if root is self:
445 new_data = new_root.config_dump()
446 else:
447 route_to_self = root.config_find_route(self)
448 new_data = cast("Self", route_to_self.get(new_root)).config_dump()
450 for key, value in new_data.items():
451 setattr(self, key, value)
453 return self
455 def _config_data_save( 1abcdefg
456 self,
457 destination: object | None = None,
458 ) -> tuple[ConfigSource[Any, Any], Data]:
459 if destination is None:
460 config_destination = self.config_source
461 else:
462 config_destination = self._validate_config_source(destination)
464 if config_destination is None:
465 msg = "Cannot save configuration (source/destination unknown)"
466 raise RuntimeError(msg)
468 root = self.config_root
469 processor = self.config_processor
471 if root is self:
472 new_data = self.config_dump()
473 else:
474 # Construct a new configuration instance.
475 # Respect `__class__` attribute: root might be a proxy, e.g. from proxyvars.
476 new_root = root.__class__(**processor.get_processed_data())
477 routes = root.config_find_routes(self)
479 for route in routes:
480 route.set(new_root, self)
482 new_data = new_root.config_dump()
484 parsed_data = processor.get_processed_data()
485 roundtrip_update_mapping(roundtrip_data=parsed_data, mergeable_data=new_data)
486 flat_new_data = parsed_data.revert_replacements()
488 data = processor.roundtrip_initial
489 config_destination.data_format.roundtrip_update_mapping(
490 roundtrip_data=data,
491 mergeable_data=flat_new_data,
492 )
493 return config_destination, data
495 def config_save(self, destination: object | None = None) -> Self: 1abcdefgh
496 """
497 Save the configuration to a given destination.
499 Parameters
500 ----------
501 destination
502 Where to save the configuration to. The argument passed is forwarded
503 to `confizen.sources.get_config_source()` which will resolve
504 the intended configuration source: for example, "abc.ini" will be resolved
505 to a TOML text file source. Keep in mind, however, that for binary formats
506 such as non-XML Plist you must specify its format type to binary, so in
507 that case just create `BinaryFileConfigSource("plist_file.plist")`.
509 """
510 config_destination, data = self._config_data_save(destination)
511 config_destination.dump(data)
512 return self
514 async def config_save_async(self, destination: object | None = None) -> Self: 1abcdefgh
515 """
516 Do the same as `config_save`, but asynchronously (no I/O blocking).
518 Parameters
519 ----------
520 destination
521 Where to save the configuration to. The argument passed is forwarded
522 to `confizen.sources.get_config_source()` which will resolve
523 the intended configuration source: for example, "abc.ini" will be resolved
524 to a TOML text file source. Keep in mind, however, that for binary formats
525 such as non-XML Plist you must specify its format type to binary, so in
526 that case just create `BinaryFileConfigSource("plist_file.plist")`.
528 """
529 config_destination, data = self._config_data_save(destination)
530 await config_destination.dump_async(data)
531 return self
533 def config_at(self, *routes: RouteLike) -> Item: 1abcdefgh
534 """Return a configuration item at the given set of routes."""
535 return Item(routes=set(map(Route, routes)), config=self)
537 def config_dump(self) -> dict[str, object]: 1abcdefgh
538 """Return a dictionary representation of the configuration."""
539 return super().model_dump()
541 def __getitem__(self, routes: RouteLike | tuple[RouteLike, ...]) -> Item: 1abcdefgh
542 """Return a configuration item at the given set of routes."""
543 if isinstance(routes, tuple):
544 return self.config_at(*routes)
545 return self.config_at(routes)
547 def __setitem__(self, item: RouteLike, value: Any) -> None: 1abcdefgh
548 """Set a configuration item at the given set of routes."""
549 self.config_at(item).config = value
551 def __init_subclass__(cls, **kwargs: Unpack[ModelConfig]) -> None: 1abcdefgh
552 """Initialize the configuration subclass."""
553 super().__init_subclass__(**cast("BaseConfigDict", kwargs)) 1abcdefgh
555 model_config: ClassVar[ModelConfig] = ModelConfig( 1abcdefgh
556 # Be lenient about forward references.
557 rebuild_on_load=True,
558 # Keep the configuration valid & fail-proof for the whole time.
559 validate_assignment=True,
560 # Make it easier to spot typos.
561 extra="forbid",
562 )
565@advance_linked_route.register(BaseConfig) 1abcdefgh
566def config_step( 1abcdefgh
567 owner: type[BaseConfig],
568 _annotation: Any,
569 step: Step[Any],
570) -> Any:
571 """Return the value of a configuration attribute."""
572 return owner.model_fields[step.key].annotation 1abcdefgh
575@dataclass 1abcdefgh
576class Item: 1abcdefgh
577 routes: set[Route] 1abcdefgh
578 config: BaseConfig 1abcdefgh
580 def __getitem__(self, item: RouteLike) -> Item: 1abcdefgh
581 return self.config.config_at(
582 *(route.enter(item) for route in self.routes),
583 )
585 def __setitem__(self, item: RouteLike, value: Any) -> None: 1abcdefgh
586 for route in self.routes:
587 route.enter(item).set(self.config, value)