Skip to content
Merged

Large diffs are not rendered by default.

2 changes: 2 additions & 0 deletions src/mountainash_settings/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
# --- Profiles + auth (2026-04-16 promotion) ---------------------------------

from .profiles import (
Adapter,
MISSING,
Missing,
ParameterSpec,
Expand Down Expand Up @@ -39,6 +40,7 @@
"get_settings_manager",

# Profiles
"Adapter",
"MISSING",
"Missing",
"ParameterSpec",
Expand Down
3 changes: 2 additions & 1 deletion src/mountainash_settings/profiles/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,12 @@

from .lookup import lookup_class_var
from .invariants import spec_invariants_for
from .profile import Profile
from .profile import Adapter, Profile
from .registry import Registry
from .spec import MISSING, Missing, ParameterSpec, ProfileSpec

__all__ = [
"Adapter",
"MISSING",
"Missing",
"ParameterSpec",
Expand Down
28 changes: 26 additions & 2 deletions src/mountainash_settings/profiles/invariants.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,32 @@ def test_parameter_names_uppercase(self, name: str, spec: t.Any) -> None:
assert p.name, f"{name}: ParameterSpec.name is empty"

def test_driver_keys_unique(self, name: str, spec: t.Any) -> None:
keys = [p.driver_key for p in spec.parameters if p.driver_key]
assert len(keys) == len(set(keys)), f"duplicate driver_key in {name}"
# Per-target output keys must be unique. A bare-string driver_key
# applies to every target; a dict driver_key applies per named
# target. (dicts are unhashable, so a set() over raw values would
# crash — resolve to per-target keys first.)
from collections import defaultdict

bare: list[str] = []
per_target: dict[t.Hashable, list[str]] = defaultdict(list)
for p in spec.parameters:
dk = p.driver_key
if not dk:
continue
if isinstance(dk, str):
bare.append(dk)
else: # dict[Hashable, str]
for target, key in dk.items():
per_target[target].append(key)

assert len(bare) == len(set(bare)), (
f"duplicate bare driver_key in {name}"
)
for target, keys in per_target.items():
combined = keys + bare # bare keys apply to every target
assert len(combined) == len(set(combined)), (
f"duplicate driver_key for target {target!r} in {name}"
)

def test_parameter_tiers_valid(self, name: str, spec: t.Any) -> None:
for p in spec.parameters:
Expand Down
122 changes: 115 additions & 7 deletions src/mountainash_settings/profiles/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,17 @@
from .lookup import lookup_class_var
from .spec import MISSING, ProfileSpec

__all__ = ["Profile"]
__all__ = ["Adapter", "Profile"]

# A target adapter composes credential/config kwargs: it receives the profile
# and the already-merged (base + driver_key renames) dict, and returns the final
# dict. Distinct from the legacy 1-arg ``__adapter__`` which owns the whole
# pipeline (see Profile docstring).
Adapter = t.Callable[["Profile", dict[str, t.Any]], dict[str, t.Any]]

# Sentinel distinguishing "no target argument passed" from an explicit ``None``
# target, so ``emit()`` can fail closed on target-scoped profiles.
_UNSET: t.Any = object()


def _resolve_spec(cls: type) -> ProfileSpec | None:
Expand Down Expand Up @@ -70,7 +80,11 @@ class Profile(MountainAshBaseSettings):
- :attr:`backend` / :attr:`profile_name` — spec name.
- :attr:`provider_type` — spec provider_type.
- :meth:`_default_kwargs` — 1:1 ``driver_key`` mappings from the spec.
- ``__adapter__`` — if set, adapter owns the output pipeline.
- :meth:`emit` — target-aware kwargs: ``driver_key`` renames →
per-target ``__adapters__`` (2-arg compose) → legacy ``__adapter__``
(1-arg, owns-pipeline) → merged dict.
- ``__adapters__`` — per-target adapter map (``{target: Adapter}``).
- ``__adapter__`` — legacy all-targets adapter; owns the output pipeline.

Public from 26.5.0. Previously named ``DescriptorProfile``.
"""
Expand All @@ -79,6 +93,7 @@ class Profile(MountainAshBaseSettings):
__adapter__: t.ClassVar[
t.Callable[["Profile"], dict[str, t.Any]] | None
] = None
__adapters__: t.ClassVar[dict[t.Hashable, "Adapter"]] = {}

@classmethod
def __pydantic_init_subclass__(cls, **kwargs: t.Any) -> None:
Expand Down Expand Up @@ -170,16 +185,36 @@ def post_init(

# --- Kwargs helpers ------------------------------------------------------

def _default_kwargs(self) -> dict[str, t.Any]:
"""Emit 1:1 ``driver_key`` mappings from the spec.
@staticmethod
def _resolve_driver_key(
driver_key: str | dict[t.Hashable, str] | None,
target: t.Hashable,
) -> str | None:
"""Resolve a param's output key for ``target``.

- ``None`` → not emitted via driver_key (adapter territory).
- bare ``str`` → that key for every target.
- ``dict`` → ``driver_key.get(target)`` (``None`` skips this param
for this target).
"""
if driver_key is None:
return None
if isinstance(driver_key, str):
return driver_key
return driver_key.get(target)

- Skips ``None`` values.
def _default_kwargs(self, target: t.Hashable = None) -> dict[str, t.Any]:
"""Emit ``driver_key`` mappings from the spec for ``target``.

- Resolves each param's key via :meth:`_resolve_driver_key`.
- Skips params whose resolved key is ``None`` and ``None`` values.
- Unwraps :class:`SecretStr` via ``.get_secret_value()``.
- Applies ``ParameterSpec.transform`` if set.
"""
out: dict[str, t.Any] = {}
for param in self.__spec__.parameters:
if param.driver_key is None:
key = self._resolve_driver_key(param.driver_key, target)
if key is None:
continue
val = getattr(self, param.name, None)
if val is None:
Expand All @@ -190,6 +225,79 @@ def _default_kwargs(self) -> dict[str, t.Any]:
val = val.get_secret_value()
if param.transform is not None:
val = param.transform(val)
out[param.driver_key] = val
out[key] = val
return out

# --- Targeting helpers ---------------------------------------------------

def _is_targeted(self) -> bool:
"""True if emission depends on a target (any per-target adapter or any
dict-scoped ``driver_key``)."""
if type(self).__adapters__:
return True
return any(
isinstance(p.driver_key, dict) for p in self.__spec__.parameters
)

def _known_targets(self) -> set[t.Hashable]:
"""Every target this profile can emit for: adapter keys ∪ dict
driver_key keys."""
targets: set[t.Hashable] = set(type(self).__adapters__)
for param in self.__spec__.parameters:
if isinstance(param.driver_key, dict):
targets.update(param.driver_key)
return targets

def _knows_target(self, target: t.Hashable) -> bool:
return target in self._known_targets()

# --- Emission ------------------------------------------------------------

def emit(
self,
target: t.Hashable = _UNSET,
*,
base: dict[str, t.Any] | None = None,
) -> dict[str, t.Any]:
"""Produce SDK kwargs for ``target``, layered onto ``base``.

Three-tier: ``driver_key`` renames, then the per-target adapter in
``__adapters__`` (2-arg compose), else the legacy ``__adapter__``
(1-arg, owns-pipeline), else the merged dict.

Fail-closed: a target-scoped profile (dict driver_keys or any
``__adapters__``) emitted with no explicit target raises rather than
silently dropping output. An unknown explicit target on such a profile
also raises.

``base`` is treated as caller-owned: only a shallow copy is taken here,
so adapters must copy-on-write any nested container they touch.
"""
if target is _UNSET:
if self._is_targeted():
raise ValueError(
f"{type(self).__name__} is target-scoped; "
f"call emit(<target>)."
)
target = None
elif self._is_targeted() and not self._knows_target(target):
# An explicit target the profile cannot serve fails closed —
# including an explicit ``None`` that is not a registered target,
# which would otherwise resolve every dict driver_key to nothing
# and emit silently. ``None`` is permitted only when it is a known
# target (``__adapters__={None: ...}`` / ``driver_key={None: ...}``).
known = sorted(self._known_targets(), key=repr)
raise ValueError(
f"{type(self).__name__} has no emission for target "
f"{target!r}; known: {known}."
)

merged = {**(base or {}), **self._default_kwargs(target)}

adapter = type(self).__adapters__.get(target)
if adapter is not None:
return adapter(self, merged)
if type(self).__adapter__ is not None:
return type(self).__adapter__(self) # legacy 1-arg owns-pipeline
return merged

11 changes: 9 additions & 2 deletions src/mountainash_settings/profiles/spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,11 @@ class ParameterSpec:
tier: ``"core"`` or ``"advanced"`` — audit-style severity tier.
default: Default value; :data:`MISSING` means the field is required.
description: Optional docstring for generated schemas / help output.
driver_key: Output-kwarg name for 1:1 mappings (e.g. ``"sslcert"``).
driver_key: Output-kwarg name for 1:1 mappings. A bare ``str`` (e.g.
``"sslcert"``) maps for every target. A ``dict[Hashable, str]``
scopes the mapping per emission target — e.g.
``{TargetFamily.PARAMIKO: "password"}`` emits only when
``emit(PARAMIKO)`` / ``_default_kwargs(PARAMIKO)`` is called.
``None`` means a domain adapter handles emission.
secret: If ``True``, wrap ``type`` as :class:`pydantic.SecretStr` and
auto-unwrap via ``.get_secret_value()`` at the kwargs boundary.
Expand All @@ -72,7 +76,10 @@ class ParameterSpec:
tier: t.Literal["core", "advanced"]
default: t.Any = MISSING
description: str = ""
driver_key: str | None = None
# dict driver_keys are unhashable; ParameterSpec is a frozen dataclass whose
# auto __hash__ would crash on a dict field. Exclude driver_key from the hash
# (it stays in __eq__) so dict-scoped specs remain hashable.
driver_key: str | dict[t.Hashable, str] | None = field(default=None, hash=False)
secret: bool = False
transform: t.Callable[[t.Any], t.Any] | None = None
validator: t.Callable[[t.Any], t.Any] | None = None
Expand Down
Loading
Loading