diff --git a/doc/api/donfig.rst b/doc/api/donfig.rst index 5c45469..0a7f0b4 100644 --- a/doc/api/donfig.rst +++ b/doc/api/donfig.rst @@ -12,6 +12,14 @@ donfig.config\_obj module :undoc-members: :show-inheritance: +donfig.typed module +------------------- + +.. automodule:: donfig.typed + :members: + :undoc-members: + :show-inheritance: + donfig.utils module ------------------- diff --git a/doc/index.rst b/doc/index.rst index e327e1d..7d8cc67 100644 --- a/doc/index.rst +++ b/doc/index.rst @@ -18,6 +18,7 @@ locations. installation configuration + typed Donfig API diff --git a/doc/typed.rst b/doc/typed.rst new file mode 100644 index 0000000..8c9d0ae --- /dev/null +++ b/doc/typed.rst @@ -0,0 +1,247 @@ +Typed Configuration +=================== + +.. currentmodule:: donfig.typed + +Donfig stores configuration as an untyped nested ``dict`` and surfaces it +through ``Config.get(key) -> Any``. That is flexible, but it means static type +checkers cannot verify configuration access: a typo in a key or a wrong +assumption about a value's type is only discovered at runtime. + +The :mod:`donfig.typed` module adds an optional, statically-typed +representation on top of donfig. Instead of a nested ``dict``, the +configuration is modeled as a tree of frozen dataclasses. This typed, +nested ``dict`` (i.e., the *schema*) is also the single source of truth for defaults. +Reads and writes still go through the familiar dotted-string API +(``config.get("a.b")``, ``config.set({"a.b": value})``), but they are +backed by immutable snapshots, and typed attribute access (``config.a.b``) +carries precise static types all the way down. + +Donfig's `typed` module provides the engine for downstream libraries to use +typed schemas, but is schema-agnostic itself. The schema, typed +attribute accessors, and per-key ``get`` overloads, live in the consuming +application. + +Donfig's YAML and environment variable ingestion is available when using the +typed module. The untyped :class:`donfig.Config` collection +machinery feeds the typed snapshot through :func:`apply_overrides`. + +Defining a schema +----------------- + +A schema is a tree of frozen dataclasses whose nodes subclass +:class:`ConfigNode`. Field defaults are the configuration defaults: + +.. code-block:: python + + # mypkg/_config.py + from dataclasses import dataclass, field + + from donfig.typed import ConfigNode + + + @dataclass(frozen=True, slots=True) + class SchedulerConfig(ConfigNode): + work_stealing: bool = False + allowed_failures: int = 5 + + + @dataclass(frozen=True, slots=True) + class MypkgConfig(ConfigNode): + logging_level: str = "info" + scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) + +:class:`ConfigNode` is a mixin that restores dict-style item access on the +dataclass nodes, so ``node["allowed_failures"]`` and dotted +``node["scheduler.allowed_failures"]`` work alongside typed attribute access +(``node.scheduler.allowed_failures``). Unknown keys raise ``KeyError``. + +Key aliases +~~~~~~~~~~~ + +Serialized key segments (from YAML files, environment variables, or +dotted-string keys) must map to dataclass field names, which are Python +identifiers. When a serialized key is not a legal identifier (.e.g, a reserved word +like ``async``, or a hyphenated key like ``work-stealing``), attach a +``__key_aliases__`` mapping (serialized segment to Python field name) to the +node: + +.. code-block:: python + + @dataclass(frozen=True, slots=True) + class SchedulerConfig(ConfigNode): + __key_aliases__ = {"work-stealing": "work_stealing"} + + work_stealing: bool = False + +Both ``config.get("scheduler.work-stealing")`` and the +``scheduler.work_stealing`` attribute then resolve to the same field. + +Creating the manager +-------------------- + +The application subclasses :class:`TypedConfigManager`, which is generic over +the root schema class. The subclass is where typed attribute accessors (and, +optionally, per-key ``get`` overloads) are defined: + +.. code-block:: python + + # mypkg/_config.py (continued) + from donfig import Config + from donfig.typed import TypedConfigManager, apply_overrides, flatten_mapping + + # Untyped donfig object: collects YAML files and MYPKG_* env vars as usual. + _ingest = Config("mypkg") + + + def _build_base() -> MypkgConfig: + return apply_overrides(MypkgConfig(), flatten_mapping(_ingest.config)) + + + class MypkgTypedConfig(TypedConfigManager[MypkgConfig]): + @property + def logging_level(self) -> str: + return self._current().logging_level + + @property + def scheduler(self) -> SchedulerConfig: + return self._current().scheduler + + + config = MypkgTypedConfig(default_factory=MypkgConfig, build_base=_build_base) + +Two callables parameterize the manager: + +``default_factory`` + Constructs the pure-defaults snapshot (typically the root schema class + itself). Exposed as the ``defaults`` property, mirroring + ``donfig.Config.defaults``. + +``build_base`` + Constructs the process-global base snapshot: the defaults overlaid with + whatever was ingested from YAML files and environment variables. + :func:`flatten_mapping` flattens the nested ``dict`` collected by + :class:`donfig.Config` into a flat dotted-key mapping, and + :func:`apply_overrides` applies that mapping to a snapshot. Unknown keys + are skipped with a warning rather than raising, so a stray environment + variable or an extra YAML key never prevents import. + +Reading configuration +--------------------- + +Both access styles read from the same state: + +.. code-block:: python + + >>> from mypkg._config import config + + >>> config.scheduler.allowed_failures # typed: int + 5 + >>> config.get("scheduler.allowed_failures") # dotted string + 5 + >>> config.get("scheduler.retries", default=3) + 3 + +``get`` raises an informative ``KeyError`` for unknown keys, suggesting the +closest match at the deepest level that resolved: + +.. code-block:: python + + >>> config.get("scheduler.allowed_failure") + Traceback (most recent call last): + ... + KeyError: "'scheduler.allowed_failure' is not a valid configuration key. Did you mean 'scheduler.allowed_failures'?" + +For interoperability with code expecting the untyped interface, ``to_dict()`` +converts the current snapshot back to a nested ``dict`` keyed by serialized +names, and ``pprint()`` prints it. + +Per-key ``get`` overloads +~~~~~~~~~~~~~~~~~~~~~~~~~ + +``TypedConfigManager.get`` returns ``Any``, like ``donfig.Config.get``. To +give the dotted-string API precise static types too, the subclass can add +``@overload`` declarations per key: + +.. code-block:: python + + from typing import Any, Literal, overload + + class MypkgTypedConfig(TypedConfigManager[MypkgConfig]): + @overload + def get(self, key: Literal["scheduler.allowed-failures"]) -> int: ... + @overload + def get(self, key: Literal["scheduler.work-stealing"]) -> bool: ... + @overload + def get(self, key: str, default: Any = ...) -> Any: ... + def get(self, key: str, default: Any = MISSING) -> Any: + return super().get(key, default) + +Setting configuration +--------------------- + +``set`` matches the semantics of ``donfig.Config.set``. Called bare, the +override is permanent, meaning it is applied immediately to the process-global base +and is visible from every thread: + +.. code-block:: python + + config.set({"scheduler.work-stealing": True}) + config.set(logging_level="debug") # keyword form + +Used as a context manager, the override is *scoped*, meaning it is isolated to the +calling context (thread or async task, via :class:`contextvars.ContextVar`) +and unwound on exit: + +.. code-block:: python + + with config.set({"scheduler.allowed-failures": 10}): + assert config.scheduler.allowed_failures == 10 + assert config.scheduler.allowed_failures == 5 + +Because snapshots are immutable frozen dataclasses, each ``set`` produces a +new snapshot; concurrent readers always see a consistent view and there is no +partially-applied state. Unlike a plain ``with config.set(...)`` on the +untyped object, the scoped form does not leak into other threads or async +tasks running concurrently. Note one asymmetry inherited from donfig's +last-writer-wins semantics: a bare ``set`` *is* visible everywhere, including +inside ``ThreadPoolExecutor`` workers, which do not copy context variables. + +``reset()`` and ``refresh()`` rebuild the base snapshot via ``build_base``, +re-reading YAML files and environment variables — the typed analog of +``donfig.Config.refresh``. + +Deprecating keys +---------------- + +Like :class:`donfig.Config`, the manager accepts a ``deprecations`` mapping +from old key name to new name, or to ``None`` for keys that were removed: + +.. code-block:: python + + config = MypkgTypedConfig( + default_factory=MypkgConfig, + build_base=_build_base, + deprecations={"sched.failures": "scheduler.allowed-failures", "old-flag": None}, + ) + +Reading or setting a renamed key warns (``DeprecationWarning`` by default — +customizable via ``deprecation_warning``) and redirects to the new key. +Setting a removed key raises; reading one honors an explicit ``default`` and +otherwise raises ``KeyError``. The exception for removed keys can be +customized via ``removed_error``. + +API +--- + +The full API is documented in the :doc:`API reference `. + +.. autosummary:: + ConfigNode + TypedConfigManager + apply_overrides + flatten_mapping + get_path + replace_path + to_nested_dict + unknown_key_error diff --git a/donfig/__init__.py b/donfig/__init__.py index 3617c49..f65c645 100644 --- a/donfig/__init__.py +++ b/donfig/__init__.py @@ -1,9 +1,11 @@ from importlib.metadata import PackageNotFoundError from importlib.metadata import version as _version -from .config_obj import Config, deserialize, serialize # noqa +from .config_obj import Config, deserialize, serialize try: __version__ = _version("donfig") except PackageNotFoundError: # pragma: no cover - source tree without metadata __version__ = "0.0.0.dev0" + +__all__ = ["Config", "deserialize", "serialize"] diff --git a/donfig/tests/test_typed.py b/donfig/tests/test_typed.py new file mode 100644 index 0000000..3744b3d --- /dev/null +++ b/donfig/tests/test_typed.py @@ -0,0 +1,447 @@ +#!/usr/bin/env python +# Copyright (c) 2018- Donfig Developers +"""Tests for the generic typed-config engine in :mod:`donfig.typed`.""" + +from __future__ import annotations + +import copy +import pickle +import sys +import threading +from collections.abc import Mapping +from concurrent.futures import ThreadPoolExecutor +from dataclasses import FrozenInstanceError, dataclass, field +from typing import Any + +import pytest + +from donfig.typed import ( + MISSING, + ConfigNode, + TypedConfigManager, + apply_overrides, + flatten_mapping, + get_path, + replace_path, + to_nested_dict, + unknown_key_error, +) + + +@dataclass(frozen=True, slots=True) +class SchedulerConfig(ConfigNode): + __key_aliases__ = {"work-stealing": "work_stealing", "async": "async_"} + + work_stealing: bool = False + allowed_failures: int = 5 + async_: bool = False + + +@dataclass(frozen=True, slots=True) +class RootConfig(ConfigNode): + logging_level: str = "info" + scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) + codecs: Mapping[str, Any] = field(default_factory=dict) + + +class TypedTestConfig(TypedConfigManager[RootConfig]): + @property + def logging_level(self) -> str: + return self._current().logging_level + + @property + def scheduler(self) -> SchedulerConfig: + return self._current().scheduler + + +def make_config(**kwargs: Any) -> TypedTestConfig: + return TypedTestConfig(default_factory=RootConfig, build_base=RootConfig, **kwargs) + + +# --- dotted-key traversal ------------------------------------------------- +def test_get_path() -> None: + cfg = RootConfig() + assert get_path(cfg, "logging_level") == "info" + assert get_path(cfg, "scheduler") is cfg.scheduler + assert get_path(cfg, "scheduler.allowed_failures") == 5 + + +def test_get_path_aliases() -> None: + cfg = RootConfig() + assert get_path(cfg, "scheduler.work-stealing") is False + assert get_path(cfg, "scheduler.async") is False + + +def test_get_path_open_mapping() -> None: + cfg = RootConfig(codecs={"blosc.shuffle": "bit"}) + # The remainder below the mapping indexes it as a single dotted key. + assert get_path(cfg, "codecs.blosc.shuffle") == "bit" + with pytest.raises(KeyError): + get_path(cfg, "codecs.zstd.level") + + +def test_get_path_unknown() -> None: + cfg = RootConfig() + with pytest.raises(KeyError): + get_path(cfg, "nope") + with pytest.raises(KeyError): + get_path(cfg, "scheduler.nope") + + +def test_replace_path() -> None: + cfg = RootConfig() + new = replace_path(cfg, "scheduler.allowed_failures", 10) + assert new.scheduler.allowed_failures == 10 + # Snapshots are immutable: the original is untouched. + assert cfg.scheduler.allowed_failures == 5 + assert replace_path(cfg, "scheduler.work-stealing", True).scheduler.work_stealing is True + + +def test_replace_path_open_mapping() -> None: + cfg = RootConfig(codecs={"a": 1}) + new = replace_path(cfg, "codecs.blosc.shuffle", "bit") + assert new.codecs == {"a": 1, "blosc.shuffle": "bit"} + assert cfg.codecs == {"a": 1} + + +def test_replace_path_unknown() -> None: + with pytest.raises(KeyError): + replace_path(RootConfig(), "scheduler.nope", 1) + + +# Ported from zarr-python#4101: a dotted key that walks past a scalar leaf must +# raise, not resolve a stray Python attribute (e.g. `str.upper`). +@pytest.mark.parametrize("key", ["logging_level.upper", "scheduler.allowed_failures.numerator"]) +def test_traversal_does_not_descend_into_scalar_attributes(key: str) -> None: + cfg = RootConfig() + with pytest.raises(KeyError): + get_path(cfg, key) + with pytest.raises(KeyError): + replace_path(cfg, key, "X") + + +def test_snapshot_is_picklable_and_deepcopyable() -> None: + cfg = replace_path(RootConfig(), "codecs.x", "pkg.X") + assert pickle.loads(pickle.dumps(cfg)) == cfg + assert copy.deepcopy(cfg) == cfg + + +def test_snapshots_are_frozen() -> None: + cfg = RootConfig() + with pytest.raises(FrozenInstanceError): + cfg.logging_level = "debug" # type: ignore[misc] + + +# --- ConfigNode item access ----------------------------------------------- +def test_config_node_getitem() -> None: + cfg = RootConfig() + assert cfg["logging_level"] == "info" + assert cfg["scheduler.allowed_failures"] == 5 + assert cfg["scheduler"] is cfg.scheduler + assert cfg.scheduler["work-stealing"] is False + with pytest.raises(KeyError): + cfg["nope"] + + +# --- typo suggestions ----------------------------------------------------- +def test_unknown_key_error_suggests_close_match() -> None: + err = unknown_key_error("scheduler.allowed_failure", RootConfig()) + assert "Did you mean 'scheduler.allowed_failures'?" in err.args[0] + + +def test_unknown_key_error_suggests_top_level() -> None: + err = unknown_key_error("logging", RootConfig()) + assert "Did you mean 'logging_level'?" in err.args[0] + + +def test_unknown_key_error_lists_valid_keys() -> None: + err = unknown_key_error("scheduler.zzz", RootConfig()) + msg = err.args[0] + assert "Valid keys under 'scheduler'" in msg + assert "allowed_failures" in msg + assert "work-stealing" in msg + + +def test_unknown_key_error_below_scalar_has_no_roster() -> None: + # The key dead-ends below a scalar leaf: nothing to suggest, no roster. + err = unknown_key_error("logging_level.x", RootConfig()) + assert err.args[0] == "'logging_level.x' is not a valid configuration key." + + +def test_unknown_key_error_resolvable_key_lists_children() -> None: + # Misuse tolerance: a key that fully resolves still yields a usable error. + err = unknown_key_error("scheduler", RootConfig()) + assert "Valid keys under 'scheduler'" in err.args[0] + # ... and one resolving to a scalar leaf has no children to list. + err = unknown_key_error("logging_level", RootConfig()) + assert err.args[0] == "'logging_level' is not a valid configuration key." + + +def test_unknown_key_error_roster_is_capped() -> None: + codecs = {f"codec{i:02d}": i for i in range(12)} + err = unknown_key_error("codecs.zzz", RootConfig(codecs=codecs)) + msg = err.args[0] + assert "codec00" in msg + assert "... (2 more)" in msg + assert "codec11" not in msg + + +# --- (de)serialization ---------------------------------------------------- +def test_to_nested_dict_uses_serialized_names() -> None: + assert to_nested_dict(RootConfig()) == { + "logging_level": "info", + "scheduler": {"work-stealing": False, "allowed_failures": 5, "async": False}, + "codecs": {}, + } + + +def test_flatten_mapping() -> None: + nested = {"a": {"b": 1, "c": {"d": 2}}, "e": 3} + assert flatten_mapping(nested) == {"a.b": 1, "a.c.d": 2, "e": 3} + + +def test_apply_overrides() -> None: + cfg = apply_overrides(RootConfig(), {"scheduler.work-stealing": True, "logging_level": "debug"}) + assert cfg.scheduler.work_stealing is True + assert cfg.logging_level == "debug" + + +def test_apply_overrides_skips_unknown_with_warning() -> None: + with pytest.warns(UserWarning, match="Unrecognized config key 'nope'"): + cfg = apply_overrides(RootConfig(), {"nope": 1, "logging_level": "debug"}) + # The unknown key is skipped; the valid one still applies. + assert cfg.logging_level == "debug" + + +def test_apply_overrides_custom_warning_category() -> None: + class IngestWarning(UserWarning): + pass + + with pytest.warns(IngestWarning): + apply_overrides(RootConfig(), {"nope": 1}, warning_category=IngestWarning) + + +# --- the manager: reads --------------------------------------------------- +def test_manager_get() -> None: + config = make_config() + assert config.get("logging_level") == "info" + assert config.get("scheduler.allowed_failures") == 5 + assert config.get("scheduler.work-stealing") is False + + +def test_manager_typed_attribute_access() -> None: + config = make_config() + assert config.logging_level == "info" + assert config.scheduler.allowed_failures == 5 + + +def test_manager_get_default() -> None: + config = make_config() + assert config.get("nope", default=3) == 3 + # An explicit None default is distinct from "no default supplied". + assert config.get("nope", default=None) is None + + +def test_manager_get_unknown_raises_with_suggestion() -> None: + config = make_config() + with pytest.raises(KeyError, match="allowed_failures"): + config.get("scheduler.allowed_failure") + + +def test_manager_to_dict_and_defaults() -> None: + config = make_config() + assert config.defaults == to_nested_dict(RootConfig()) + config.set(logging_level="debug") + assert config.to_dict()["logging_level"] == "debug" + # `defaults` still reflects the pure schema defaults. + assert config.defaults["logging_level"] == "info" + + +def test_manager_pprint(capsys: pytest.CaptureFixture[str]) -> None: + make_config().pprint() + assert "logging_level" in capsys.readouterr().out + + +# --- the manager: writes -------------------------------------------------- +def test_manager_bare_set_is_permanent() -> None: + config = make_config() + config.set({"scheduler.allowed_failures": 7}) + config.set(logging_level="debug") # keyword form + assert config.scheduler.allowed_failures == 7 + assert config.logging_level == "debug" + + +def test_manager_bare_set_visible_across_threads() -> None: + config = make_config() + config.set({"scheduler.allowed_failures": 7}) + with ThreadPoolExecutor(1) as pool: + assert pool.submit(config.get, "scheduler.allowed_failures").result() == 7 + + +def test_manager_set_unknown_key_raises() -> None: + config = make_config() + with pytest.raises(KeyError, match="allowed_failures"): + config.set({"scheduler.allowed_failure": 1}) + + +def test_manager_get_and_set_reject_descending_into_scalar() -> None: + config = make_config() + with pytest.raises(KeyError): + config.get("logging_level.upper") + with pytest.raises(KeyError): + config.set({"logging_level.upper": "X"}) + + +def test_manager_permanent_set_cross_thread_last_writer_wins() -> None: + # A permanent `set` from any thread updates the shared global base, so a + # later permanent `set` in another thread is visible everywhere. + config = make_config() + config.set({"scheduler.allowed_failures": 1}) + worker = threading.Thread(target=lambda: config.set({"scheduler.allowed_failures": 999})) + worker.start() + worker.join() + assert config.get("scheduler.allowed_failures") == 999 + + +def test_manager_concurrent_sets_to_distinct_keys_all_survive() -> None: + # `set` rebuilds the whole snapshot from `_base`; the manager locks that + # read-modify-write so concurrent sets to distinct keys don't lose updates. + # A tiny switch interval forces preemption inside the critical section. + config = make_config() + n = 32 + barrier = threading.Barrier(n) + + def worker(i: int) -> None: + barrier.wait() + config.set({f"codecs.k{i}": i}) + + old_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-6) + try: + with ThreadPoolExecutor(max_workers=n) as pool: + list(pool.map(worker, range(n))) + finally: + sys.setswitchinterval(old_interval) + codecs = config.get("codecs") + assert all(codecs.get(f"k{i}") == i for i in range(n)) + + +def test_manager_scoped_set_unwinds() -> None: + config = make_config() + with config.set({"scheduler.allowed_failures": 10}): + assert config.scheduler.allowed_failures == 10 + assert config.scheduler.allowed_failures == 5 + + +def test_manager_scoped_set_nests() -> None: + config = make_config() + with config.set(logging_level="debug"): + with config.set(logging_level="warning"): + assert config.logging_level == "warning" + assert config.logging_level == "debug" + assert config.logging_level == "info" + + +def test_manager_scoped_set_reverts_open_mapping_changes() -> None: + # A scoped `set` that *adds* a new mapping key removes it again on block + # exit, while a pre-existing key it overrode is restored to its prior value. + def build_base() -> RootConfig: + return RootConfig(codecs={"blosc": "pkg.Blosc"}) + + config = TypedTestConfig(default_factory=RootConfig, build_base=build_base) + with config.set({"codecs.new_codec": "pkg.New", "codecs.blosc": "pkg.Override"}): + assert config.get("codecs.new_codec") == "pkg.New" + assert config.get("codecs.blosc") == "pkg.Override" + codecs = config.get("codecs") + assert "new_codec" not in codecs + assert codecs["blosc"] == "pkg.Blosc" + + +def test_manager_scoped_set_is_context_local() -> None: + config = make_config() + with config.set({"scheduler.allowed_failures": 10}): + # A fresh thread has a fresh context: it sees the base, not the overlay. + with ThreadPoolExecutor(1) as pool: + assert pool.submit(config.get, "scheduler.allowed_failures").result() == 5 + + +def test_manager_bare_set_inside_scope_does_not_leak_overlay() -> None: + config = make_config() + with config.set(logging_level="debug"): + config.set({"scheduler.allowed_failures": 9}) + # The permanent write is masked while the scope overlay is active. + assert config.scheduler.allowed_failures == 5 + # On exit the permanent write survives, but the scoped override does not. + assert config.scheduler.allowed_failures == 9 + assert config.logging_level == "info" + + +def test_manager_update() -> None: + config = make_config() + config.update({"logging_level": "debug"}) + assert config.logging_level == "debug" + + +# --- the manager: lifecycle ----------------------------------------------- +def test_manager_reset_discards_overrides() -> None: + config = make_config() + config.set(logging_level="debug") + config.reset() + assert config.logging_level == "info" + + +def test_manager_refresh_rebuilds_base() -> None: + state = {"level": "info"} + + def build_base() -> RootConfig: + return RootConfig(logging_level=state["level"]) + + config = TypedTestConfig(default_factory=RootConfig, build_base=build_base) + assert config.logging_level == "info" + state["level"] = "warning" + config.refresh() + assert config.logging_level == "warning" + + +# --- the manager: deprecations -------------------------------------------- +def test_manager_renamed_key_redirects_with_warning() -> None: + config = make_config(deprecations={"sched.failures": "scheduler.allowed_failures"}) + with pytest.warns(DeprecationWarning, match="renamed to 'scheduler.allowed_failures'"): + config.set({"sched.failures": 3}) + assert config.scheduler.allowed_failures == 3 + with pytest.warns(DeprecationWarning): + assert config.get("sched.failures") == 3 + + +def test_manager_removed_key() -> None: + config = make_config(deprecations={"old-flag": None}) + with pytest.raises(ValueError, match="'old-flag' has been removed"): + config.set({"old-flag": 1}) + # Reads honour an explicit default (even None), else raise KeyError. + assert config.get("old-flag", default=2) == 2 + assert config.get("old-flag", default=None) is None + with pytest.raises(KeyError): + config.get("old-flag") + + +def test_manager_custom_deprecation_hooks() -> None: + class MyDeprecationWarning(UserWarning): + pass + + def removed_error(key: str) -> Exception: + return RuntimeError(f"gone: {key}") + + config = make_config( + deprecations={"gone": None, "old": "logging_level"}, + removed_error=removed_error, + deprecation_warning=MyDeprecationWarning, + ) + with pytest.raises(RuntimeError, match="gone"): + config.set({"gone": 1}) + with pytest.warns(MyDeprecationWarning): + config.set({"old": "debug"}) + assert config.logging_level == "debug" + + +def test_missing_sentinel_is_not_none() -> None: + assert MISSING is not None diff --git a/donfig/typed.py b/donfig/typed.py new file mode 100644 index 0000000..e186fde --- /dev/null +++ b/donfig/typed.py @@ -0,0 +1,448 @@ +""" +Generic engine for statically-typed, dataclass-backed configuration. + +donfig historically stored configuration as an untyped nested ``dict`` and +surfaced it through ``Config.get(key) -> Any``. This module adds an optional, +*typed* representation that any donfig consumer can adopt without giving up +donfig's env-var / YAML ingestion: model the configuration as a tree of frozen +dataclasses (the schema, and the single source of truth for defaults) and drive +it through the familiar dotted-string API, but backed by immutable snapshots and +a context-local overlay for scoped ``with`` blocks. + +To adopt it an application: + +* defines its schema as frozen ``@dataclass`` nodes subclassing `ConfigNode`. + Attach ``__key_aliases__`` (serialized-segment -> Python-field-name) to a node + when a serialized key is not a legal identifier, e.g. ``{"async": "async_"}``; +* subclasses `TypedConfigManager`, passing ``default_factory`` (the root schema + class) and ``build_base`` (defaults overlaid with env/YAML ingest, typically + via :func:`apply_overrides` fed from ``donfig.Config(...).config``); +* adds per-key ``get`` overloads and typed attribute ``@property`` accessors so + that both ``config.get("a.b")`` and ``config.a.b`` carry precise static types. + +Everything in this module is schema-agnostic; the schema, the overloads, and the +typed accessors live in the consuming application. +""" + +from __future__ import annotations + +import difflib +import threading +import warnings +from collections.abc import Callable, Mapping +from contextvars import ContextVar, Token +from dataclasses import fields, is_dataclass, replace +from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, overload + +if TYPE_CHECKING: + from typing_extensions import Self + +#: A frozen-dataclass config snapshot type (the application's root schema). +ConfigT = TypeVar("ConfigT") + +__all__ = [ + "MISSING", + "ConfigNode", + "TypedConfigManager", + "apply_overrides", + "flatten_mapping", + "get_path", + "replace_path", + "to_nested_dict", + "unknown_key_error", +] + +#: Sentinel distinguishing "no default supplied" from an explicit ``None`` default. +MISSING: Any = object() + +_ROSTER_LIMIT = 10 + + +# --- key-alias helpers ---------------------------------------------------- +def _key_aliases(obj: object) -> Mapping[str, str]: + """Serialized-segment -> Python-field-name overrides for a config node.""" + return getattr(type(obj), "__key_aliases__", {}) + + +def _serialized_names(obj: object) -> Mapping[str, str]: + """Python-field-name -> serialized-segment (the reverse of ``__key_aliases__``).""" + return {field_name: serialized for serialized, field_name in _key_aliases(obj).items()} + + +def _resolve_field(obj: object, segment: str) -> str: + """Translate a serialized key segment to the dataclass field name.""" + return _key_aliases(obj).get(segment, segment) + + +# --- dotted-key traversal ------------------------------------------------- +def get_path(cfg: object, key: str) -> object: + """Read a dotted-string key from a frozen-dataclass config snapshot. + + Raises + ------ + KeyError + If the key does not resolve to a value. + """ + obj: object = cfg + segments = key.split(".") + for i, segment in enumerate(segments): + if isinstance(obj, Mapping): + # remaining segments index into an open mapping (e.g. codecs.*) + remainder = ".".join(segments[i:]) + try: + return obj[remainder] + except KeyError: + raise KeyError(key) from None + # A prior segment resolved to a scalar leaf, but the key has more + # segments — descend no further. Without this guard, `hasattr` would + # match ordinary Python attributes/methods (e.g. `logging_level.upper` + # returning `str.upper`) instead of raising for the invalid key. + if not is_dataclass(obj): + raise KeyError(key) + field_name = _resolve_field(obj, segment) + if field_name not in {f.name for f in fields(obj)}: + raise KeyError(key) + obj = getattr(obj, field_name) + return obj + + +def replace_path(cfg: ConfigT, key: str, value: object) -> ConfigT: + """Return a new snapshot with the dotted-string key set to ``value``. + + The return type mirrors the input: replacing a value produces a new snapshot of + the same schema type, so callers keep their precise static type without a cast. + """ + segments = key.split(".") + return _replace_recursive(cfg, segments, value, key) # type: ignore[return-value] + + +# `obj: Any` is load-bearing here: the function dispatches dynamically between a +# `Mapping` (open subtree) and a dataclass instance, and `dataclasses.replace` +# requires a dataclass-typed argument that `object` would reject. +def _replace_recursive(obj: Any, segments: list[str], value: object, key: str) -> object: + segment = segments[0] + if isinstance(obj, Mapping): + remainder = ".".join(segments) + return {**obj, remainder: value} + # See the scalar-leaf guard in `get_path`: never descend past a non-node. + if not is_dataclass(obj): + raise KeyError(key) + field_name = _resolve_field(obj, segment) + if field_name not in {f.name for f in fields(obj)}: + raise KeyError(key) + # `is_dataclass` narrows `obj` to `... | type[...]`, which `replace` rejects; + # at runtime `obj` is always a dataclass *instance* here, so re-widen to Any. + node: Any = obj + if len(segments) == 1: + return replace(node, **{field_name: value}) + child = getattr(node, field_name) + new_child = _replace_recursive(child, segments[1:], value, key) + return replace(node, **{field_name: new_child}) + + +# --- typo suggestions ----------------------------------------------------- +def _children(obj: object) -> list[str]: + """Return the immediate child key names of a config node (else an empty list).""" + if isinstance(obj, Mapping): + return list(obj) + if is_dataclass(obj): + names = _serialized_names(obj) + return [names.get(f.name, f.name) for f in fields(obj)] + return [] + + +def _resolve_for_suggestion(cfg: object, key: str) -> tuple[str, list[str], str]: + """Walk ``key`` as far as it resolves. + + Returns the deepest resolvable dotted prefix, that node's child key names, and + the first segment that failed to resolve (the remainder is treated as a single + key once an open mapping is reached). + """ + obj: object = cfg + prefix = "" + segments = key.split(".") + for i, segment in enumerate(segments): + if isinstance(obj, Mapping): + # the remainder indexes into an open mapping as a single key + return prefix, _children(obj), ".".join(segments[i:]) + if not is_dataclass(obj): + # the key descends below a scalar leaf: nothing to suggest there + return prefix, [], segment + field_name = _resolve_field(obj, segment) + if field_name not in {f.name for f in fields(obj)}: + return prefix, _children(obj), segment + obj = getattr(obj, field_name) + prefix = f"{prefix}.{segment}" if prefix else segment + return prefix, _children(obj), "" + + +def unknown_key_error(key: str, cfg: object) -> KeyError: + """Build a `KeyError` for an unknown config key. + + Resolves ``key`` to the deepest valid level, then suggests the closest child + key there if one is similar enough; otherwise lists the available keys at that + level (capped at ``_ROSTER_LIMIT``). + """ + msg = f"{key!r} is not a valid configuration key." + prefix, children, failed = _resolve_for_suggestion(cfg, key) + matches = difflib.get_close_matches(failed, children, n=1) if failed != "" else [] + if len(matches) > 0: + suggestion = f"{prefix}.{matches[0]}" if prefix != "" else matches[0] + return KeyError(f"{msg} Did you mean {suggestion!r}?") + if len(children) > 0: + shown = sorted(children) + roster = ", ".join(shown[:_ROSTER_LIMIT]) + if len(shown) > _ROSTER_LIMIT: + roster += f", ... ({len(shown) - _ROSTER_LIMIT} more)" + where = f" under {prefix!r}" if prefix != "" else "" + msg = f"{msg} Valid keys{where}: {roster}." + return KeyError(msg) + + +# --- (de)serialization ---------------------------------------------------- +def to_nested_dict(cfg: object) -> dict[str, Any]: + """Convert a config snapshot to a nested dict keyed by serialized names. + + Returns a heterogeneous, JSON-like tree (nested dicts and scalars) that + callers navigate by key, so `Any` values are appropriate here. + """ + + # `obj: Any` is also load-bearing: `dataclasses.fields` requires a + # dataclass-typed argument that `object` would reject. + def convert(obj: Any) -> Any: + if isinstance(obj, Mapping): + return dict(obj) + if hasattr(type(obj), "__dataclass_fields__"): + names = _serialized_names(obj) + out: dict[str, Any] = {} + for f in fields(obj): + out[names.get(f.name, f.name)] = convert(getattr(obj, f.name)) + return out + return obj + + return convert(cfg) # type: ignore[no-any-return] + + +def flatten_mapping(data: Mapping[str, object], prefix: str = "") -> dict[str, object]: + """Flatten a nested mapping into a single dotted-key mapping.""" + out: dict[str, object] = {} + for k, v in data.items(): + key = f"{prefix}{k}" if not prefix else f"{prefix}.{k}" + if isinstance(v, Mapping): + out.update(flatten_mapping(v, key)) + else: + out[key] = v + return out + + +def apply_overrides( + cfg: ConfigT, + overrides: Mapping[str, object], + *, + warning_category: type[Warning] = UserWarning, +) -> ConfigT: + """Apply a flat dotted-key override map to a snapshot. + + Unknown keys are skipped with a warning rather than raising, so a stray + environment variable or extra YAML key never prevents import. + """ + for key, value in overrides.items(): + try: + cfg = replace_path(cfg, key, value) + except KeyError: + warnings.warn( + f"Unrecognized config key {key!r} from environment or YAML — ignoring.", + warning_category, + stacklevel=2, + ) + return cfg + + +# --- item-access mixin ---------------------------------------------------- +class ConfigNode: + """Mixin giving frozen config dataclasses dict-style item access. + + A typed config returns dataclass instances for subtrees; this mixin restores + subscripting (``node["order"]`` and dotted ``node["a.b"]``) alongside typed + attribute access (``node.order``), raising `KeyError` for unknown keys. + + ``__slots__ = ()`` keeps subclasses fully slotted (no ``__dict__``). + """ + + __slots__ = () + + def __getitem__(self, key: str) -> object: + return get_path(self, key) + + +# --- scoped-set context manager ------------------------------------------- +class _ConfigSet: + """Context manager returned by ``TypedConfigManager.set``. + + ``set`` applies the override immediately to the process-global base, so a bare + ``config.set(...)`` is permanent and visible from every thread (matching + donfig's last-writer-wins semantics, including inside ``ThreadPoolExecutor`` + workers, which do not copy context variables). + + Using the result as a ``with`` block *promotes* the override to a + context-local scope: ``__enter__`` undoes the global apply and re-applies the + new snapshot through a `ContextVar`, so the change is isolated to the calling + context (thread / async task) and unwound on ``__exit__``. + """ + + def __init__(self, manager: TypedConfigManager[Any], prev_base: object, new: object) -> None: + self._manager = manager + self._prev_base = prev_base + self._new = new + self._token: Token[Any] | None = None + + def __enter__(self) -> Self: + self._token = self._manager._enter_scope(self._prev_base, self._new) + return self + + def __exit__(self, *exc: object) -> None: + if self._token is not None: + self._manager._exit_scope(self._token) + + +def _default_removed_error(key: str) -> Exception: + return ValueError(f"Configuration key {key!r} has been removed and no longer has any effect.") + + +# --- the manager ---------------------------------------------------------- +class TypedConfigManager(Generic[ConfigT]): + """Schema-agnostic base for a typed, donfig-compatible configuration object. + + Generic over ``ConfigT``, the application's root schema dataclass. Holds + immutable ``ConfigT`` snapshots and resolves reads through a process-global + base plus a context-local overlay, so `_current` (and therefore the subclass's + typed attribute properties) carries the precise schema type. Subclasses add + per-key ``get`` overloads and typed attribute properties over `_current`. + """ + + def __init__( + self, + *, + default_factory: Callable[[], ConfigT], + build_base: Callable[[], ConfigT], + deprecations: Mapping[str, str | None] | None = None, + removed_error: Callable[[str], Exception] = _default_removed_error, + deprecation_warning: type[Warning] = DeprecationWarning, + ) -> None: + self._default_factory = default_factory + self._build_base = build_base + self._deprecations: Mapping[str, str | None] = dict(deprecations or {}) + self._removed_error = removed_error + self._deprecation_warning = deprecation_warning + self._base: ConfigT = build_base() + self._scope: ContextVar[ConfigT] = ContextVar("typed_config_scope") + # Serializes read-modify-write of the process-global `_base` so + # concurrent `set`s to different keys don't lose updates (each rebuilds + # a whole immutable snapshot from `_base`). + self._lock = threading.Lock() + + # --- state resolution ------------------------------------------------- + def _current(self) -> ConfigT: + return self._scope.get(self._base) + + def _enter_scope(self, prev_base: ConfigT, new: ConfigT) -> Token[ConfigT]: + with self._lock: + self._base = prev_base + return self._scope.set(new) + + def _exit_scope(self, token: Token[ConfigT]) -> None: + self._scope.reset(token) + + # --- string API ------------------------------------------------------- + def get(self, key: str, default: Any = MISSING) -> Any: + resolved = self._apply_deprecation(key, raise_on_removed=False) + if resolved is None: + # Key was removed; treat as absent — honour the caller's default. + if default is MISSING: + raise KeyError(key) + return default + current = self._current() + try: + return get_path(current, resolved) + except KeyError: + if default is MISSING: + raise unknown_key_error(key, current) from None + return default + + def set(self, updates: Mapping[str, object] | None = None, **kwargs: object) -> _ConfigSet: + """Apply one or more config overrides (permanent, or scoped as a ``with``).""" + all_updates: dict[str, object] = {} + if updates: + all_updates.update(updates) + all_updates.update(kwargs) + with self._lock: + prev_base = self._base + # `scoped` layers on the current view (any active `with` overlay); it is + # what a `with config.set(...)` pins as its context-local scope. `permanent` + # layers on the global base, so a bare `set` nested inside a `with` block + # does not leak that block's overlay into the base. + scoped = self._current() + permanent = prev_base + for key, value in all_updates.items(): + resolved = self._apply_deprecation(key, raise_on_removed=True) + try: + scoped = replace_path(scoped, resolved, value) + permanent = replace_path(permanent, resolved, value) + except KeyError: + raise unknown_key_error(key, permanent) from None + self._base = permanent + return _ConfigSet(self, prev_base, scoped) + + # --- lifecycle -------------------------------------------------------- + def reset(self) -> None: + # Rebuild outside the lock (`build_base` may read env/YAML) and swap + # atomically under it. + new_base = self._build_base() + with self._lock: + self._base = new_base + + def refresh(self) -> None: + self.reset() + + # --- compat / introspection ------------------------------------------ + @property + def defaults(self) -> dict[str, Any]: + return to_nested_dict(self._default_factory()) + + def to_dict(self) -> dict[str, Any]: + return to_nested_dict(self._current()) + + def update(self, updates: Mapping[str, object]) -> None: + self.set(updates) + + def pprint(self) -> None: + import pprint as _pp + + _pp.pprint(self.to_dict()) + + # --- deprecations ----------------------------------------------------- + @overload + def _apply_deprecation(self, key: str, *, raise_on_removed: Literal[True]) -> str: ... + @overload + def _apply_deprecation(self, key: str, *, raise_on_removed: Literal[False]) -> str | None: ... + def _apply_deprecation(self, key: str, *, raise_on_removed: bool) -> str | None: + """Resolve a possibly-deprecated config key to its canonical name. + + Returns the (possibly redirected) key, or ``None`` when the key was + removed and ``raise_on_removed`` is ``False`` (so the caller can honour a + default). Raises ``removed_error(key)`` when removed and + ``raise_on_removed`` is ``True``. + """ + if key not in self._deprecations: + return key + new_key = self._deprecations[key] + if new_key is None: + if raise_on_removed: + raise self._removed_error(key) + return None + warnings.warn( + f"Configuration key {key!r} has been renamed to {new_key!r}.", + self._deprecation_warning, + stacklevel=3, + ) + return new_key