Coverage for src/configuraptor/core.py: 100%
253 statements
« prev ^ index » next coverage.py v7.2.7, created at 2026-05-01 14:18 +0200
1"""
2Contains most of the loading logic.
3"""
5import dataclasses as dc
6import io
7import os
8import typing
9import warnings
10from pathlib import Path
11from typing import Any, Type
13import requests
14from dotenv import dotenv_values as _dotenv_values
15from dotenv import find_dotenv
17from . import loaders
18from .abs import DEFAULT_ENV_SETTING, AnyType, C, T, T_data, Type_C, UseEnvSetting
19from .alias import Alias, has_alias
20from .binary_config import BinaryConfig
21from .errors import (
22 ConfigErrorCouldNotConvert,
23 ConfigErrorInvalidType,
24 ConfigErrorMissingKey,
25 FailedToLoad,
26)
27from .helpers import (
28 all_annotations,
29 camel_to_snake,
30 check_type,
31 dataclass_field,
32 expand_env_vars_into_toml_values,
33 find_pyproject_toml,
34 is_custom_class,
35 is_optional,
36 is_parameterized,
37 is_union,
38)
39from .postpone import Postponed
40from .type_converters import CONVERTERS
43def _data_for_nested_key(key: str, raw: dict[str, typing.Any]) -> dict[str, typing.Any]:
44 """
45 If a key contains a dot, traverse the raw dict until the right key was found.
47 Example:
48 key = some.nested.key
49 raw = {"some": {"nested": {"key": {"with": "data"}}}}
50 -> {"with": "data"}
51 """
52 parts = key.split(".")
53 while parts:
54 key = parts.pop(0)
55 if key not in raw:
56 return {}
58 raw = raw[key]
60 return raw
63def _guess_key(clsname: str) -> str:
64 """
65 If no key is manually defined for `load_into`, \
66 the class' name is converted to snake_case to use as the default key.
67 """
68 return camel_to_snake(clsname)
71def _from_mock_url(url: str) -> str:
72 """
73 Pytest only: when starting a url with mock:// it is expected to just be json afterwards.
74 """
75 return url.removeprefix("mock://")
78def guess_filetype_for_url(url: str, response: requests.Response = None) -> str:
79 """
80 Based on the url (which may have an extension) and the requests response \
81 (which may have a content-type), try to guess the right filetype (-> loader, e.g. json or yaml).
83 Falls back to JSON if none can be found.
84 """
85 url = url.split("?")[0]
86 if url_extension := os.path.splitext(url)[1].lower():
87 return url_extension.strip(".")
89 if response and (content_type_header := response.headers.get("content-type", "").split(";")[0].strip()):
90 content_type = content_type_header.split("/")[-1]
91 if content_type != "plain":
92 return content_type
94 # If both methods fail, default to JSON
95 return "json"
98def from_url(url: str, _dummy: bool = False) -> tuple[io.BytesIO, str]:
99 """
100 Load data as bytes into a file-like object and return the file type.
102 This can be used by __load_data:
103 > loader = loaders.get(filetype)
104 > # dev/null exists but always returns b''
105 > data = loader(contents, Path("/dev/null"))
106 """
107 if url.startswith("mock://"):
108 data = _from_mock_url(url)
109 resp = None
110 elif _dummy:
111 resp = None
112 data = "{}"
113 else:
114 ssl_verify = os.getenv("SSL_VERIFY", "1") == "1"
116 resp = requests.get(url, timeout=10, verify=ssl_verify)
117 data = resp.text
119 filetype = guess_filetype_for_url(url, resp)
120 return io.BytesIO(data.encode()), filetype
123def dotenv_values() -> dict[str, str | None]:
124 """Wrapper around dotenv.dotenv_values that uses .env in cwd."""
125 return _dotenv_values(dotenv_path=find_dotenv(usecwd=True))
128def apply_env(data: dict[str, typing.Any], use_env: UseEnvSetting) -> None:
129 """
130 Apply the desired env-setting logic on data.
131 """
132 match use_env:
133 case "yes":
134 env = dotenv_values() | os.environ
135 case "inverse":
136 env = os.environ | dotenv_values()
137 case "dotenv":
138 env = dotenv_values()
139 case "environ":
140 env = {**os.environ}
141 case _: # pragma: no cover
142 return
144 expand_env_vars_into_toml_values(data, env)
147def _load_data(
148 data: T_data,
149 key: str = None,
150 classname: str = None,
151 lower_keys: bool = False,
152 allow_types: tuple[type, ...] = (dict,),
153 strict: bool = False,
154 use_env: UseEnvSetting = DEFAULT_ENV_SETTING,
155) -> dict[str, typing.Any]:
156 """
157 Tries to load the right data from a filename/path or dict, based on a manual key or a classname.
159 E.g. class Tool will be mapped to key tool.
160 It also deals with nested keys (tool.extra -> {"tool": {"extra": ...}}
161 """
162 if isinstance(data, bytes):
163 # instantly return, don't modify
164 # bytes as inputs -> bytes as output
165 # but since `T_data` is re-used, that's kind of hard to type for mypy.
166 return data # type: ignore
168 if isinstance(data, list):
169 if not data:
170 raise ValueError("Empty list passed!")
172 final_data: dict[str, typing.Any] = {}
173 for source in data:
174 final_data |= load_data(
175 source,
176 key=key,
177 classname=classname,
178 lower_keys=True,
179 allow_types=allow_types,
180 strict=strict,
181 use_env=use_env,
182 )
184 return final_data
186 if isinstance(data, str):
187 if data.startswith(("http://", "https://", "mock://")):
188 contents, filetype = from_url(data)
190 loader = loaders.get(filetype)
191 # dev/null exists but always returns b''
192 data = loader(contents, Path("/dev/null"))
193 else:
194 data = Path(data)
196 if isinstance(data, Path):
197 with data.open("rb") as f:
198 loader = loaders.get(data.suffix or data.name)
199 data = loader(f, data.resolve())
201 if not data:
202 return {}
204 if key is None:
205 # try to guess key by grabbing the first one or using the class name
206 if len(data) == 1:
207 key = next(iter(data.keys()))
208 elif classname is not None:
209 key = _guess_key(classname)
211 if key:
212 data = _data_for_nested_key(key, data)
214 if not data:
215 raise ValueError("No data found!")
217 if not isinstance(data, allow_types):
218 raise ValueError(f"Data should be one of {allow_types} but it is {type(data)}!")
220 if lower_keys and isinstance(data, dict):
221 data = {k.lower(): v for k, v in data.items()}
223 if use_env != "no" and isinstance(data, dict):
224 apply_env(data, use_env)
226 return typing.cast(dict[str, typing.Any], data)
229def load_data(
230 data: T_data,
231 key: str = None,
232 classname: str = None,
233 lower_keys: bool = False,
234 allow_types: tuple[type, ...] = (dict,),
235 strict: bool = False,
236 use_env: UseEnvSetting = DEFAULT_ENV_SETTING,
237) -> dict[str, typing.Any]:
238 """
239 Wrapper around __load_data that retries with key="" if anything goes wrong.
240 """
241 if data is None:
242 # try to load pyproject.toml
243 data = find_pyproject_toml()
245 try:
246 return _load_data(
247 data,
248 key,
249 classname,
250 lower_keys=lower_keys,
251 allow_types=allow_types,
252 strict=strict,
253 use_env=use_env,
254 )
255 except Exception as e:
256 # sourcery skip: remove-unnecessary-else, simplify-empty-collection-comparison, swap-if-else-branches
257 # @sourcery: `key != ""` is NOT the same as `not key`
258 if key != "":
259 # try again with key ""
260 return load_data(
261 data,
262 "",
263 classname,
264 lower_keys=lower_keys,
265 allow_types=allow_types,
266 strict=strict,
267 use_env=use_env,
268 )
269 elif strict:
270 raise FailedToLoad(data) from e
271 else:
272 # e.g. if settings are to be loaded via a URL that is unavailable or returns invalid json
273 warnings.warn(f"Data ('{data!r}') could not be loaded", source=e, category=UserWarning)
274 return {}
277F = typing.TypeVar("F")
280def convert_between(from_value: F, from_type: Type[F], to_type: Type[T]) -> T:
281 """
282 Convert a value between types.
283 """
284 if converter := CONVERTERS.get((from_type, to_type)):
285 return typing.cast(T, converter(from_value))
287 # default: just convert type:
288 return to_type(from_value) # type: ignore
291def check_and_convert_type(value: Any, _type: Type[T], convert_types: bool, key: str = "variable") -> T:
292 """
293 Checks if the given value matches the specified type. If it does, the value is returned as is.
295 Args:
296 value (Any): The value to be checked and potentially converted.
297 _type (Type[T]): The expected type for the value.
298 convert_types (bool): If True, allows type conversion if the types do not match.
299 key (str, optional): The name or key associated with the variable (used in error messages).
300 Defaults to "variable".
302 Returns:
303 T: The value, potentially converted to the expected type.
305 Raises:
306 ConfigErrorInvalidType: If the type does not match, and type conversion is not allowed.
307 ConfigErrorCouldNotConvert: If type conversion fails.
308 """
309 if check_type(value, _type):
310 # type matches
311 return value
313 if isinstance(value, Alias):
314 if is_optional(_type):
315 return typing.cast(T, None)
316 else:
317 # unresolved alias, error should've already been thrown for parent but lets do it again:
318 raise ConfigErrorInvalidType(value.to, value=value, expected_type=_type)
320 if not convert_types:
321 # type does not match and should not be converted
322 raise ConfigErrorInvalidType(key, value=value, expected_type=_type)
324 # else: type does not match, try to convert it
325 try:
326 return convert_between(value, type(value), _type)
327 except (TypeError, ValueError) as e:
328 raise ConfigErrorCouldNotConvert(type(value), _type, value) from e
331def ensure_types(
332 data: dict[str, T],
333 annotations: dict[str, type[T]],
334 convert_types: bool = False,
335) -> dict[str, T | None]:
336 """
337 Make sure all values in 'data' are in line with the ones stored in 'annotations'.
339 If an annotated key in missing from data, it will be filled with None for convenience.
341 TODO: python 3.11 exception groups to throw multiple errors at once!
342 """
343 # custom object to use instead of None, since typing.Optional can be None!
344 # cast to T to make mypy happy
345 notfound = typing.cast(T, object())
347 final: dict[str, T | None] = {}
348 for key, _type in annotations.items():
349 compare = data.get(key, notfound)
350 if compare is notfound: # pragma: nocover
351 warnings.warn("This should not happen since `load_recursive` already fills `data` based on `annotations`")
352 # skip!
353 continue
355 if isinstance(compare, Postponed):
356 # don't do anything with this item!
357 continue
359 if isinstance(compare, Alias):
360 related_data = data.get(compare.to, notfound)
361 if related_data is not notfound:
362 if isinstance(related_data, Postponed):
363 # also continue alias for postponed items
364 continue
366 # original key set, update alias
367 compare = related_data
369 compare = check_and_convert_type(compare, _type, convert_types, key)
371 final[key] = compare
373 return final
376def convert_key(key: str) -> str:
377 """
378 Replaces '-' and '.' in keys with '_' so it can be mapped to the Config properties.
379 """
380 return key.replace("-", "_").replace(".", "_")
383def convert_config(items: dict[str, T]) -> dict[str, T]:
384 """
385 Converts the config dict (from toml) or 'overwrites' dict in two ways.
387 1. removes any items where the value is None, since in that case the default should be used;
388 2. replaces '-' and '.' in keys with '_' so it can be mapped to the Config properties.
389 """
390 return {convert_key(k): v for k, v in items.items() if v is not None}
393def load_recursive(
394 cls: AnyType,
395 data: dict[str, T],
396 annotations: dict[str, AnyType],
397 convert_types: bool = False,
398) -> dict[str, T]:
399 """
400 For all annotations (recursively gathered from parents with `all_annotations`), \
401 try to resolve the tree of annotations.
403 Uses `load_into_recurse`, not itself directly.
405 Example:
406 class First:
407 key: str
409 class Second:
410 other: First
412 # step 1
413 cls = Second
414 data = {"second": {"other": {"key": "anything"}}}
415 annotations: {"other": First}
417 # step 1.5
418 data = {"other": {"key": "anything"}
419 annotations: {"other": First}
421 # step 2
422 cls = First
423 data = {"key": "anything"}
424 annotations: {"key": str}
427 TODO: python 3.11 exception groups to throw multiple errors at once!
428 """
429 updated = {}
431 for _key, _type in annotations.items():
432 if _key in data:
433 value: typing.Any = data[_key] # value can change so define it as any instead of T
434 if is_parameterized(_type):
435 origin = typing.get_origin(_type)
436 arguments = typing.get_args(_type)
437 if origin is list and arguments and is_custom_class(arguments[0]):
438 subtype = arguments[0]
439 value = [_load_into_recurse(subtype, subvalue, convert_types=convert_types) for subvalue in value]
441 elif origin is dict and arguments and is_custom_class(arguments[1]):
442 # e.g. dict[str, Point]
443 subkeytype, subvaluetype = arguments
444 # subkey(type) is not a custom class, so don't try to convert it:
445 value = {
446 subkey: _load_into_recurse(subvaluetype, subvalue, convert_types=convert_types)
447 for subkey, subvalue in value.items()
448 }
449 # elif origin is dict:
450 # keep data the same
451 elif is_union(_type) and arguments:
452 for arg in arguments:
453 if is_custom_class(arg):
454 value = _load_into_recurse(arg, value, convert_types=convert_types)
456 elif is_custom_class(_type):
457 # type must be C (custom class) at this point; includes dataclass but not optional[cls]
458 value = _load_into_recurse(
459 # make mypy and pycharm happy by telling it _type is of type C...
460 # actually just passing _type as first arg!
461 typing.cast(Type_C[typing.Any], _type),
462 value,
463 convert_types=convert_types,
464 )
466 # else: normal value, don't change
468 elif value := has_alias(cls, _key, data):
469 # value updated by alias
470 ...
471 elif _key in cls.__dict__:
472 # property has default, use that instead.
473 value = cls.__dict__[_key]
474 elif is_optional(_type):
475 # type is optional and not found in __dict__ -> default is None
476 value = None
477 elif dc.is_dataclass(cls) and (field := dataclass_field(cls, _key)) and field.default_factory is not dc.MISSING:
478 # could have a default factory
479 # todo: do something with field.default?
480 value = field.default_factory()
481 elif is_custom_class(_type) and isinstance(_type, type) and issubclass(_type, Defaultable):
482 value = _type.default()
483 else:
484 raise ConfigErrorMissingKey(_key, cls, _type)
486 updated[_key] = value
488 return updated
491def check_and_convert_data(
492 cls: typing.Type[C],
493 data: dict[str, typing.Any],
494 _except: typing.Iterable[str],
495 strict: bool = True,
496 convert_types: bool = False,
497) -> dict[str, typing.Any]:
498 """
499 Based on class annotations, this prepares the data for `load_into_recurse`.
501 1. convert config-keys to python compatible config_keys
502 2. loads custom class type annotations with the same logic (see also `load_recursive`)
503 3. ensures the annotated types match the actual types after loading the config file.
504 """
505 annotations = all_annotations(cls, _except=_except)
507 to_load = convert_config(data)
508 to_load = load_recursive(cls, to_load, annotations, convert_types=convert_types)
510 if strict:
511 to_load = ensure_types(to_load, annotations, convert_types=convert_types)
513 return to_load
516T_init_list = list[typing.Any]
517T_init_dict = dict[str, typing.Any]
518T_init = tuple[T_init_list, T_init_dict] | T_init_list | T_init_dict | None
521@typing.no_type_check # (mypy doesn't understand 'match' fully yet)
522def _split_init(init: T_init) -> tuple[T_init_list, T_init_dict]:
523 """
524 Accept a tuple, a dict or a list of (arg, kwarg), {kwargs: ...}, [args] respectively and turn them all into a tuple.
525 """
526 if not init:
527 return [], {}
529 args: T_init_list = []
530 kwargs: T_init_dict = {}
531 match init:
532 case (args, kwargs):
533 return args, kwargs
534 case [*args]:
535 return args, {}
536 case {**kwargs}:
537 return [], kwargs
538 case _:
539 raise ValueError("Init must be either a tuple of list and dict, a list or a dict.")
542def _load_into_recurse(
543 cls: typing.Type[C],
544 data: dict[str, typing.Any] | bytes,
545 init: T_init = None,
546 strict: bool = True,
547 convert_types: bool = False,
548) -> C:
549 """
550 Loads an instance of `cls` filled with `data`.
552 Uses `load_recursive` to load any fillable annotated properties (see that method for an example).
553 `init` can be used to optionally pass extra __init__ arguments. \
554 NOTE: This will overwrite a config key with the same name!
555 """
556 init_args, init_kwargs = _split_init(init)
558 if isinstance(data, bytes) or issubclass(cls, BinaryConfig):
559 if not isinstance(data, (bytes, dict)): # pragma: no cover
560 raise NotImplementedError("BinaryConfig can only deal with `bytes` or a dict of bytes as input.")
561 elif not issubclass(cls, BinaryConfig): # pragma: no cover
562 raise NotImplementedError("Only BinaryConfig can be used with `bytes` (or a dict of bytes) as input.")
564 inst = typing.cast(C, cls._parse_into(data))
565 elif dc.is_dataclass(cls):
566 to_load = check_and_convert_data(cls, data, init_kwargs.keys(), strict=strict, convert_types=convert_types)
567 if init:
568 raise ValueError("Init is not allowed for dataclasses!")
570 # ensure mypy inst is an instance of the cls type (and not a fictuous `DataclassInstance`)
571 inst = typing.cast(C, cls(**to_load))
572 elif isinstance(data, cls):
573 # already the right type! (e.g. Pathlib)
574 inst = typing.cast(C, data)
575 else:
576 inst = cls(*init_args, **init_kwargs)
577 to_load = check_and_convert_data(cls, data, inst.__dict__.keys(), strict=strict, convert_types=convert_types)
578 inst.__dict__.update(**to_load)
580 return inst
583def _load_into_instance(
584 inst: C,
585 cls: typing.Type[C],
586 data: dict[str, typing.Any],
587 init: T_init = None,
588 strict: bool = True,
589 convert_types: bool = False,
590) -> C:
591 """
592 Similar to `load_into_recurse` but uses an existing instance of a class (so after __init__) \
593 and thus does not support init.
595 """
596 if init is not None:
597 raise ValueError("Can not init an existing instance!")
599 existing_data = inst.__dict__
601 to_load = check_and_convert_data(
602 cls,
603 data,
604 _except=existing_data.keys(),
605 strict=strict,
606 convert_types=convert_types,
607 )
609 inst.__dict__.update(**to_load)
611 return inst
614def load_into_class(
615 cls: typing.Type[C],
616 data: T_data,
617 /,
618 key: str = None,
619 init: T_init = None,
620 strict: bool = True,
621 lower_keys: bool = False,
622 convert_types: bool = False,
623 use_env: UseEnvSetting = DEFAULT_ENV_SETTING,
624) -> C:
625 """
626 Shortcut for _load_data + load_into_recurse.
627 """
628 allow_types = (dict, bytes) if issubclass(cls, BinaryConfig) else (dict,)
629 to_load = load_data(
630 data,
631 key,
632 cls.__name__,
633 lower_keys=lower_keys,
634 allow_types=allow_types,
635 strict=strict,
636 use_env=use_env,
637 )
638 return _load_into_recurse(cls, to_load, init=init, strict=strict, convert_types=convert_types)
641def load_into_instance(
642 inst: C,
643 data: T_data,
644 /,
645 key: str = None,
646 init: T_init = None,
647 strict: bool = True,
648 lower_keys: bool = False,
649 convert_types: bool = False,
650 use_env: UseEnvSetting = DEFAULT_ENV_SETTING,
651) -> C:
652 """
653 Shortcut for _load_data + load_into_existing.
654 """
655 cls = inst.__class__
656 allow_types = (dict, bytes) if issubclass(cls, BinaryConfig) else (dict,)
657 to_load = load_data(
658 data,
659 key,
660 cls.__name__,
661 lower_keys=lower_keys,
662 allow_types=allow_types,
663 strict=strict,
664 use_env=use_env,
665 )
666 return _load_into_instance(inst, cls, to_load, init=init, strict=strict, convert_types=convert_types)
669def load_into(
670 cls: typing.Type[C],
671 data: T_data = None,
672 /,
673 key: str = None,
674 init: T_init = None,
675 strict: bool = True,
676 lower_keys: bool = False,
677 convert_types: bool = False,
678 use_env: UseEnvSetting = DEFAULT_ENV_SETTING,
679) -> C:
680 """
681 Load your config into a class (instance).
683 Supports both a class or an instance as first argument, but that's hard to explain to mypy, so officially only
684 classes are supported, and if you want to `load_into` an instance, you should use `load_into_instance`.
686 Args:
687 cls: either a class or an existing instance of that class.
688 data: can be a dictionary or a path to a file to load (as pathlib.Path or str)
689 key: optional (nested) dictionary key to load data from (e.g. 'tool.su6.specific')
690 init: optional data to pass to your cls' __init__ method (only if cls is not an instance already)
691 strict: enable type checks or allow anything?
692 lower_keys: should the config keys be lowercased? (for .env)
693 convert_types: should the types be converted to the annotated type if not yet matching? (for .env)
694 use_env: Controls how ${VAR} placeholders are resolved.
695 Determines which sources are consulted and in what order:
697 - "yes" (default): OS environment → .env
698 - "inverse": .env → OS environment
699 - "dotenv": .env only
700 - "environ": OS environment only
701 - "no": no interpolation
702 """
703 result: C
705 if not isinstance(cls, type):
706 # would not be supported according to mypy, but you can still load_into(instance)
707 result = load_into_instance(
708 cls,
709 data,
710 key=key,
711 init=init,
712 strict=strict,
713 lower_keys=lower_keys,
714 convert_types=convert_types,
715 use_env=use_env,
716 )
717 else:
718 # get instance of cls()
719 result = load_into_class(
720 cls,
721 data,
722 key=key,
723 init=init,
724 strict=strict,
725 lower_keys=lower_keys,
726 convert_types=convert_types,
727 use_env=use_env,
728 )
730 post_init = getattr(result, "__post_init__", None)
731 if callable(post_init) and not dc.is_dataclass(result):
732 post_init()
734 return result
737class Defaultable:
738 """
739 Explicit opt-in for classes that can construct a default instance.
740 """
742 @classmethod
743 def default(cls) -> typing.Self:
744 """
745 Return a default instance of `cls`.
746 """
747 return load_into(cls, {})