"""Typed configuration field descriptors and view field types.
Each concrete class here is a `ConfigField` specialization. `Alias` and
`Mirror` are descriptor classes declared on views or configs to wire fields
together.
"""
import datetime
import json
import numbers
import pathlib
import re
from ..refs import FieldRef
from ..utils import walk, walk_set
try:
import click as _click
except ImportError:
_click = None
from .config_field import ConfigField
__all__ = [
"ConfigField",
"Any",
"String",
"Int",
"Float",
"Scalar",
"Bool",
"Options",
"MultiOptions",
"Path",
"Seed",
"Range",
"List",
"Dict",
"Date",
"Time",
"DateTime",
"Alias",
"Mirror",
]
[docs]
class Any(ConfigField):
"""A field that accepts any value without validation.
Use this as an explicit escape hatch when validation is intentionally
deferred to `Config.validate` or handled externally. The ``Any`` name
signals intent: the author has consciously opted out of field-level
constraints.
Parameters
----------
default_value : `object` or `callable`
Default value or lazy factory.
doc : `str`
Documentation.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
"""
[docs]
class Options(ConfigField):
"""A field whose value must be one of a fixed set of choices.
Parameters
----------
options : `tuple` or `list`
Allowed values. The first element is used as the default unless
``default_value`` is provided.
doc : `str`
Documentation.
default_value : `object`, optional
Default value. Must be a member of ``options``. Defaults to
``options[0]``.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
ValueError
If the value being set is not in ``options``.
Examples
--------
>>> from cfx import Config
>>> from cfx.types import Options
>>> class BaseConfig(Config):
... mode = Options(("fast", "balanced", "thorough"), "Processing mode")
>>> cfg = BaseConfig()
>>> cfg.mode
'fast'
>>> cfg.mode = "thorough"
>>> cfg.mode = "turbo"
Traceback (most recent call last):
...
ValueError: Expected 'turbo' to be one of ('fast', 'balanced', 'thorough')
"""
def __init__(
self,
options,
doc,
default_value=None,
static=False,
env=None,
transient=None,
):
self.options = options
defaultval = options[0] if default_value is None else default_value
super().__init__(
defaultval,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def validate(self, value): # noqa: D102
if value not in self.options:
raise ValueError(f"Expected {value!r} to be one of {self.options}")
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["choices"] = list(self.options)
return kwargs
[docs]
def to_click_option(self, name, prefix=""): # noqa: D102
if _click is None:
raise ImportError(
"click is required for click_options(). "
"Install it with: pip install click"
)
hyphenated = name.replace("_", "-")
flag = f"--{prefix}.{hyphenated}" if prefix else f"--{hyphenated}"
param_name = f"{prefix.replace('.', '__')}__{name}" if prefix else name
return _click.option(
flag,
param_name,
type=_click.Choice(list(self.options)),
default=None,
help=self.doc,
show_default=False,
)
[docs]
class MultiOptions(ConfigField):
"""A field whose value must be a subset of a fixed set of choices.
Parameters
----------
options : `tuple` or `list`
Full set of allowed values.
doc : `str`
Documentation.
default_value : `set` or `frozenset`, optional
Default selection. Defaults to an empty set.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `set` or `frozenset`.
ValueError
If the value contains elements not in ``options``.
Examples
--------
>>> from cfx import Config
>>> from cfx.types import MultiOptions
>>> class PipelineConfig(Config):
... steps = MultiOptions(
... ("preprocess", "detect", "cluster", "output"),
... "Pipeline steps to run",
... default_value={"preprocess", "detect"},
... )
"""
def __init__(
self,
options,
doc,
default_value=None,
static=False,
env=None,
transient=None,
):
self.options = set(options)
defaultval = set() if default_value is None else default_value
super().__init__(
defaultval,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def normalize(self, value): # noqa: D102
# Coerce list/tuple -> set: YAML deserializes sequences as lists,
# and click's multiple=True returns a tuple.
if isinstance(value, (list, tuple)):
return set(value)
return value
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, (set, frozenset)):
raise TypeError(f"Expected a set, got {type(value).__name__!r}")
extras = set(value) - self.options
if extras:
raise ValueError(
f"Expected {self.options!r}, got {extras!r} instead"
)
[docs]
def serialize(self, value): # noqa: D102
return sorted(value, key=str)
[docs]
def deserialize(self, value): # noqa: D102
if isinstance(value, (list, tuple)):
return set(value)
return value
[docs]
def from_string(self, s): # noqa: D102
return {item.strip() for item in s.split(",") if item.strip()}
[docs]
def to_string(self, value): # noqa: D102
return "{" + ", ".join(sorted(str(x) for x in value)) + "}"
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["nargs"] = "+"
kwargs["choices"] = list(self.options)
kwargs["type"] = str
return kwargs
[docs]
def to_click_option(self, name, prefix=""): # noqa: D102
if _click is None:
raise ImportError(
"click is required for click_options(). "
"Install it with: pip install click"
)
hyphenated = name.replace("_", "-")
flag = f"--{prefix}.{hyphenated}" if prefix else f"--{hyphenated}"
param_name = f"{prefix.replace('.', '__')}__{name}" if prefix else name
return _click.option(
flag,
param_name,
type=_click.Choice(list(self.options)),
multiple=True,
default=None,
help=self.doc,
show_default=False,
)
[docs]
class String(ConfigField):
"""A field that validates string values.
Parameters
----------
default_value : `str` or `callable`
Default string value or lazy factory returning a `str`.
doc : `str`
Documentation.
min_length : `int`, optional
Minimum allowed string length. Default is ``0``.
max_length : `int` or `None`, optional
Maximum allowed string length. `None` means no upper limit.
Default is `None`.
pattern : `str` or `None`, optional
Regex pattern the value must fully match (``re.fullmatch``).
Default is `None`.
predicate : `callable` or `None`, optional
Additional validation callable that accepts a `str` and returns
`True` if the value is valid. Not expressible in JSON Schema; use
``pattern=`` when a regex is sufficient. Default is `None`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `str`.
ValueError
If the value does not satisfy the length, pattern, or predicate
constraints.
"""
def __init__(
self,
default_value,
doc,
min_length=0,
max_length=None,
pattern=None,
predicate=None,
static=False,
env=None,
transient=None,
):
self.min_length = min_length
self.max_length = max_length
self.pattern = pattern
self.predicate = predicate
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, str):
raise TypeError(f"Expected a str, got {type(value).__name__!r}")
if len(value) < self.min_length:
raise ValueError(
f"Expected string of at least {self.min_length} characters, "
f"got {len(value)}"
)
if self.max_length is not None and len(value) > self.max_length:
raise ValueError(
f"Expected string of at most {self.max_length} characters, "
f"got {len(value)}"
)
if self.pattern is not None and not re.fullmatch(self.pattern, value):
raise ValueError(
f"Value {value!r} does not match pattern {self.pattern!r}"
)
if self.predicate is not None and not self.predicate(value):
raise ValueError(
f"Value {value!r} failed predicate {self.predicate!r}"
)
[docs]
class Scalar(ConfigField):
"""A field that validates numeric (int or float) values.
Use this when both integers and floats are acceptable. For fields that
must be strictly one numeric type, prefer `Int` or `Float`.
Parameters
----------
default_value : `numbers.Number` or `callable`
Default numeric value or lazy factory returning a number.
doc : `str`
Documentation.
ge : `numbers.Number` or `None`, optional
Minimum allowed value (inclusive, ``>=``). Default is `None`.
gt : `numbers.Number` or `None`, optional
Strict lower bound (exclusive, ``>``). Default is `None`.
le : `numbers.Number` or `None`, optional
Maximum allowed value (inclusive, ``<=``). Default is `None`.
lt : `numbers.Number` or `None`, optional
Strict upper bound (exclusive, ``<``). Default is `None`.
multiple_of : `numbers.Number` or `None`, optional
Value must be a multiple of this. Default is `None`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `numbers.Number`.
ValueError
If the value violates any bound or ``multiple_of`` constraint.
"""
def __init__(
self,
default_value,
doc,
ge=None,
gt=None,
le=None,
lt=None,
multiple_of=None,
static=False,
env=None,
transient=None,
):
self.ge = ge
self.gt = gt
self.le = le
self.lt = lt
self.multiple_of = multiple_of
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def from_string(self, s): # noqa: D102
try:
return int(s)
except ValueError:
pass
try:
return float(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as a number (env var {self.env!r})"
) from err
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, numbers.Number):
raise TypeError(
f"Expected a numeric value, got {type(value).__name__!r}"
)
if self.gt is not None and not (value > self.gt):
raise ValueError(f"Expected value > {self.gt!r}, got {value!r}")
if self.ge is not None and not (value >= self.ge):
raise ValueError(f"Expected value >= {self.ge!r}, got {value!r}")
if self.lt is not None and not (value < self.lt):
raise ValueError(f"Expected value < {self.lt!r}, got {value!r}")
if self.le is not None and not (value <= self.le):
raise ValueError(f"Expected value <= {self.le!r}, got {value!r}")
if self.multiple_of is not None and value % self.multiple_of != 0:
raise ValueError(
f"Expected multiple of {self.multiple_of!r}, got {value!r}"
)
[docs]
class Int(ConfigField):
"""A field that validates integer values.
Parameters
----------
default_value : `int` or `callable`
Default integer value or lazy factory returning an `int`.
doc : `str`
Documentation.
ge : `int` or `None`, optional
Minimum allowed value (inclusive, ``>=``). Default is `None`.
gt : `int` or `None`, optional
Strict lower bound (exclusive, ``>``). Default is `None`.
le : `int` or `None`, optional
Maximum allowed value (inclusive, ``<=``). Default is `None`.
lt : `int` or `None`, optional
Strict upper bound (exclusive, ``<``). Default is `None`.
multiple_of : `int` or `None`, optional
Value must be a multiple of this. Default is `None`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
env : `str` or `None`, optional
Environment variable name. See `ConfigField`.
Raises
------
TypeError
If the value is not an `int`.
ValueError
If the value violates any bound or ``multiple_of`` constraint.
"""
def __init__(
self,
default_value,
doc,
ge=None,
gt=None,
le=None,
lt=None,
multiple_of=None,
static=False,
env=None,
transient=None,
):
self.ge = ge
self.gt = gt
self.le = le
self.lt = lt
self.multiple_of = multiple_of
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def from_string(self, s): # noqa: D102
try:
return int(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as int (env var {self.env!r})"
) from err
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, int) or isinstance(value, bool):
raise TypeError(f"Expected an int, got {type(value).__name__!r}")
if self.gt is not None and not (value > self.gt):
raise ValueError(f"Expected value > {self.gt!r}, got {value!r}")
if self.ge is not None and not (value >= self.ge):
raise ValueError(f"Expected value >= {self.ge!r}, got {value!r}")
if self.lt is not None and not (value < self.lt):
raise ValueError(f"Expected value < {self.lt!r}, got {value!r}")
if self.le is not None and not (value <= self.le):
raise ValueError(f"Expected value <= {self.le!r}, got {value!r}")
if self.multiple_of is not None and value % self.multiple_of != 0:
raise ValueError(
f"Expected multiple of {self.multiple_of!r}, got {value!r}"
)
[docs]
class Float(ConfigField):
"""A field that validates floating-point values.
Parameters
----------
default_value : `float` or `callable`
Default float value or lazy factory returning a `float`.
doc : `str`
Documentation.
ge : `float` or `None`, optional
Minimum allowed value (inclusive, ``>=``). Default is `None`.
gt : `float` or `None`, optional
Strict lower bound (exclusive, ``>``). Default is `None`.
le : `float` or `None`, optional
Maximum allowed value (inclusive, ``<=``). Default is `None`.
lt : `float` or `None`, optional
Strict upper bound (exclusive, ``<``). Default is `None`.
multiple_of : `float` or `None`, optional
Value must be a multiple of this. Default is `None`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
env : `str` or `None`, optional
Environment variable name. See `ConfigField`.
Raises
------
TypeError
If the value is not a `float` or `int` (ints are coerced).
ValueError
If the value violates any bound or ``multiple_of`` constraint.
"""
def __init__(
self,
default_value,
doc,
ge=None,
gt=None,
le=None,
lt=None,
multiple_of=None,
static=False,
env=None,
transient=None,
):
self.ge = ge
self.gt = gt
self.le = le
self.lt = lt
self.multiple_of = multiple_of
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def from_string(self, s): # noqa: D102
try:
return float(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as float (env var {self.env!r})"
) from err
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, (float, int)) or isinstance(value, bool):
raise TypeError(f"Expected a float, got {type(value).__name__!r}")
if self.gt is not None and not (value > self.gt):
raise ValueError(f"Expected value > {self.gt!r}, got {value!r}")
if self.ge is not None and not (value >= self.ge):
raise ValueError(f"Expected value >= {self.ge!r}, got {value!r}")
if self.lt is not None and not (value < self.lt):
raise ValueError(f"Expected value < {self.lt!r}, got {value!r}")
if self.le is not None and not (value <= self.le):
raise ValueError(f"Expected value <= {self.le!r}, got {value!r}")
if self.multiple_of is not None and value % self.multiple_of != 0:
raise ValueError(
f"Expected multiple of {self.multiple_of!r}, got {value!r}"
)
[docs]
class Bool(ConfigField):
"""A field that validates boolean values.
Parameters
----------
default_value : `bool` or `callable`
Default boolean value or lazy factory returning a `bool`.
doc : `str`
Documentation.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `bool`.
"""
[docs]
def from_string(self, s): # noqa: D102
if s.lower() in ("1", "true", "yes", "on"):
return True
if s.lower() in ("0", "false", "no", "off"):
return False
raise ValueError(
f"Cannot parse {s!r} as bool (env var {self.env!r}). "
f"Use 1/0, true/false, yes/no, or on/off."
)
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, bool):
raise TypeError(f"Expected a bool, got {type(value).__name__!r}")
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
import argparse
kwargs = super().to_argparse_kwargs(name, prefix)
del kwargs["type"]
kwargs["action"] = argparse.BooleanOptionalAction
return kwargs
[docs]
def to_click_option(self, name, prefix=""): # noqa: D102
if _click is None:
raise ImportError(
"click is required for click_options(). "
"Install it with: pip install click"
)
hyphenated = name.replace("_", "-")
if prefix:
flag_pair = f"--{prefix}.{hyphenated}/--{prefix}.no-{hyphenated}"
param_name = f"{prefix.replace('.', '__')}__{name}"
else:
flag_pair = f"--{hyphenated}/--no-{hyphenated}"
param_name = name
return _click.option(
flag_pair,
param_name,
default=None,
help=self.doc,
show_default=False,
)
[docs]
class Path(ConfigField):
"""A field that coerces and validates filesystem paths.
Accepts `str` or `pathlib.Path` values and stores them as
`pathlib.Path` objects.
Parameters
----------
default_value : `str`, `pathlib.Path`, or `callable`
Default path value or lazy factory. Strings are coerced to
`pathlib.Path` on assignment.
doc : `str`
Documentation.
must_exist : `bool`, optional
If `True`, raises `ValueError` when the path does not exist on
the filesystem at the time of assignment. Default is `False`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value cannot be coerced to a `pathlib.Path`.
ValueError
If ``must_exist`` is `True` and the path does not exist.
"""
def __init__(
self,
default_value,
doc,
must_exist=False,
static=False,
env=None,
transient=None,
):
self.must_exist = must_exist
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def normalize(self, value): # noqa: D102
# Coerce str/os.PathLike -> pathlib.Path. Let validate() handle
# any TypeError if the value is truly not path-like.
try:
return pathlib.Path(value)
except TypeError:
return value
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, pathlib.Path):
raise TypeError(f"Expected a Path, got {type(value).__name__!r}")
if self.must_exist and not value.exists():
raise ValueError(f"Path does not exist: {value!r}")
[docs]
def serialize(self, value): # noqa: D102
# as_posix() ensures forward slashes on all platforms; str() would
# produce backslashes on Windows, breaking cross-platform round-trips.
return value.as_posix()
[docs]
def deserialize(self, value): # noqa: D102
return pathlib.Path(value)
[docs]
def from_string(self, s): # noqa: D102
return pathlib.Path(s)
[docs]
def to_string(self, value): # noqa: D102
return str(value)
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "PATH"
return kwargs
[docs]
class Seed(ConfigField):
"""A field for random-number-generator seeds.
Accepts an `int` or `None`. A value of `None` signals that the seed
should be chosen randomly at runtime, distinct from any specific integer
seed including zero.
Parameters
----------
default_value : `int`, `None`, or `callable`
Default seed value or lazy factory.
doc : `str`
Documentation.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not an `int` or `None`.
"""
[docs]
def from_string(self, s): # noqa: D102
if s.lower() in ("none", "null", ""):
return None
try:
return int(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as seed (env var {self.env!r}). "
f"Use an integer or 'none'."
) from err
[docs]
def validate(self, value): # noqa: D102
if value is not None and not isinstance(value, int):
raise TypeError(
f"Expected an int or None for a seed value, "
f"got {type(value).__name__!r}"
)
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "INT|none"
return kwargs
[docs]
class Range(ConfigField):
"""A field for a linear (min, max) numeric range.
Stores a two-element tuple ``(min, max)`` and validates that
``min < max``. This field is intentionally limited to linear ranges.
Cyclical or angular ranges (e.g. 350 deg to 10 deg crossing zero) are out
of scope because their validation depends on domain-specific conventions.
Parameters
----------
default_value : `tuple[numbers.Number, numbers.Number]` or `callable`
Default ``(min, max)`` tuple or lazy factory.
doc : `str`
Documentation.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a two-element sequence of numbers.
ValueError
If ``min >= max``.
"""
[docs]
def normalize(self, value): # noqa: D102
# Coerce list -> tuple: YAML deserializes sequences as lists.
if isinstance(value, list):
return tuple(value)
return value
[docs]
def serialize(self, value): # noqa: D102
return list(value)
[docs]
def deserialize(self, value): # noqa: D102
if isinstance(value, list):
return tuple(value)
return value
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "MIN,MAX"
return kwargs
[docs]
def from_string(self, s): # noqa: D102
parts = s.split(",")
if len(parts) != 2:
raise ValueError(
f"Cannot parse {s!r} as range (env var {self.env!r}). "
f"Expected 'min,max' (e.g. '0.0,1.0')."
)
def _num(x):
x = x.strip()
try:
return int(x)
except ValueError:
try:
return float(x)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as range (env var {self.env!r}). "
f"Expected 'min,max' (e.g. '0.0,1.0')."
) from err
return (_num(parts[0]), _num(parts[1]))
[docs]
def validate(self, value): # noqa: D102
try:
lo, hi = value
except (TypeError, ValueError) as err:
raise TypeError(
f"Expected a (min, max) pair, got {value!r}"
) from err
if not (
isinstance(lo, numbers.Number) and isinstance(hi, numbers.Number)
): # noqa: E501
raise TypeError(
f"Range bounds must be numeric, got {lo!r} and {hi!r}"
) # noqa: E501
if lo >= hi:
raise ValueError(f"Expected min < max, got ({lo!r}, {hi!r})")
[docs]
class List(ConfigField):
"""A field that validates list values.
Parameters
----------
default_value : `list` or `callable`
Default list value or lazy factory returning a `list`.
doc : `str`
Documentation.
element_type : `type` or `None`, optional
If provided, every element of the list must be an instance of
this type. Default is `None` (no element-level type check).
min_length : `int` or `None`, optional
Minimum number of elements. Default is `None`.
max_length : `int` or `None`, optional
Maximum number of elements. Default is `None`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `list`, or contains elements of the wrong
type when ``element_type`` is set.
ValueError
If the list length is outside [``min_length``, ``max_length``].
"""
def __init__(
self,
default_value,
doc,
element_type=None,
min_length=None,
max_length=None,
static=False,
env=None,
transient=None,
):
self.element_type = element_type
self.min_length = min_length
self.max_length = max_length
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def normalize(self, value): # noqa: D102
# Coerce tuple -> list: click's multiple=True returns a tuple.
if isinstance(value, tuple):
return list(value)
return value
[docs]
def serialize(self, value): # noqa: D102
return list(value)
[docs]
def deserialize(self, value): # noqa: D102
return list(value) if not isinstance(value, list) else value
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["nargs"] = "+"
kwargs["type"] = (
self.element_type if self.element_type is not None else str
)
return kwargs
[docs]
def to_click_option(self, name, prefix=""): # noqa: D102
if _click is None:
raise ImportError(
"click is required for click_options(). "
"Install it with: pip install click"
)
hyphenated = name.replace("_", "-")
flag = f"--{prefix}.{hyphenated}" if prefix else f"--{hyphenated}"
param_name = f"{prefix.replace('.', '__')}__{name}" if prefix else name
et = self.element_type if self.element_type is not None else str
return _click.option(
flag,
param_name,
type=et,
multiple=True,
default=None,
help=self.doc,
show_default=False,
)
[docs]
def from_string(self, s): # noqa: D102
try:
result = json.loads(s)
if isinstance(result, list):
return result
except (json.JSONDecodeError, ValueError):
pass
return [item.strip() for item in s.split(",") if item.strip()]
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, list):
raise TypeError(f"Expected a list, got {type(value).__name__!r}")
if self.min_length is not None and len(value) < self.min_length:
raise ValueError(
f"Expected at least {self.min_length} elements, "
"got {len(value)} instead."
)
if self.max_length is not None and len(value) > self.max_length:
raise ValueError(
f"Expected at most {self.max_length} elements, "
"got {len(value)} instead"
)
if self.element_type is not None:
bad = [el for el in value if not isinstance(el, self.element_type)]
if bad:
raise TypeError(
f"All elements must be {self.element_type.__name__!r}, "
f"found {bad!r}"
)
[docs]
class Dict(ConfigField):
"""A field that validates dict values.
Accepts any `dict`. Use this for free-form sub-structure that is too
loosely typed to warrant a nested `Config`, but too structured to be
a `String`.
Parameters
----------
default_value : `dict` or `callable`
Default dict value or lazy factory returning a `dict`.
doc : `str`
Documentation.
min_length : `int` or `None`, optional
Minimum number of keys. Default is `None`.
max_length : `int` or `None`, optional
Maximum number of keys. Default is `None`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `dict`.
ValueError
If the number of keys violates ``min_length`` or ``max_length``.
"""
def __init__(
self,
default_value,
doc,
min_length=None,
max_length=None,
static=False,
env=None,
transient=None,
):
self.min_length = min_length
self.max_length = max_length
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def from_string(self, s): # noqa: D102
try:
return json.loads(s)
except (json.JSONDecodeError, ValueError) as err:
raise ValueError(
f"Cannot parse {s!r} as JSON dict (env var {self.env!r})"
) from err
[docs]
def to_string(self, value): # noqa: D102
return json.dumps(value)
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, dict):
raise TypeError(f"Expected a dict, got {type(value).__name__!r}")
if self.min_length is not None and len(value) < self.min_length:
raise ValueError(
f"Expected at least {self.min_length} keys, got {len(value)}"
)
if self.max_length is not None and len(value) > self.max_length:
raise ValueError(
f"Expected at most {self.max_length} keys, got {len(value)}"
)
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "JSON"
return kwargs
[docs]
class Date(ConfigField):
"""A field that validates `datetime.date` values.
Parameters
----------
default_value : `datetime.date` or `callable`
Default date value or lazy factory returning a `datetime.date`.
doc : `str`
Documentation.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `datetime.date`.
"""
[docs]
def normalize(self, value): # noqa: D102
# Coerce ISO string -> datetime.date for round-trips through to_dict.
if isinstance(value, str):
return datetime.date.fromisoformat(value)
return value
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, datetime.date):
raise TypeError(
f"Expected a datetime.date, got {type(value).__name__!r}"
)
[docs]
def serialize(self, value): # noqa: D102
return value.isoformat()
[docs]
def deserialize(self, value): # noqa: D102
if isinstance(value, str):
return datetime.date.fromisoformat(value)
return value
[docs]
def from_string(self, s): # noqa: D102
try:
return datetime.date.fromisoformat(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as ISO date (env var {self.env!r})"
) from err
[docs]
def to_string(self, value): # noqa: D102
return value.isoformat()
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "YYYY-MM-DD"
return kwargs
[docs]
class Time(ConfigField):
"""A field that validates `datetime.time` values.
Parameters
----------
default_value : `datetime.time` or `callable`
Default time value or lazy factory returning a `datetime.time`.
doc : `str`
Documentation.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `datetime.time`.
"""
[docs]
def normalize(self, value): # noqa: D102
# Coerce ISO string -> datetime.time: to_dict serializes time as an
# ISO string (YAML has no native time type), so from_dict round-trips
# through here.
if isinstance(value, str):
return datetime.time.fromisoformat(value)
return value
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, datetime.time):
raise TypeError(
f"Expected a datetime.time, got {type(value).__name__!r}"
)
[docs]
def serialize(self, value): # noqa: D102
return value.isoformat()
[docs]
def deserialize(self, value): # noqa: D102
if isinstance(value, str):
return datetime.time.fromisoformat(value)
return value
[docs]
def from_string(self, s): # noqa: D102
try:
return datetime.time.fromisoformat(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as ISO time (env var {self.env!r})"
) from err
[docs]
def to_string(self, value): # noqa: D102
return value.isoformat()
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "HH:MM:SS"
return kwargs
[docs]
class DateTime(ConfigField):
"""A field that validates `datetime.datetime` values.
Parameters
----------
default_value : `datetime.datetime` or `callable`
Default datetime value or lazy factory returning a
`datetime.datetime`.
doc : `str`
Documentation.
tz_aware : `bool`, optional
If `True`, the value must have a non-`None` ``tzinfo``. Default
is `False`.
static : `bool`, optional
If `True`, the value is frozen at class-definition time.
Default is `False`.
Raises
------
TypeError
If the value is not a `datetime.datetime`.
ValueError
If ``tz_aware=True`` and the datetime has no timezone info.
"""
def __init__(
self,
default_value,
doc,
tz_aware=False,
static=False,
env=None,
transient=None,
):
self.tz_aware = tz_aware
super().__init__(
default_value,
doc,
static=static,
env=env,
transient=transient,
)
[docs]
def normalize(self, value): # noqa: D102
if isinstance(value, str):
return datetime.datetime.fromisoformat(value)
return value
[docs]
def validate(self, value): # noqa: D102
if not isinstance(value, datetime.datetime):
raise TypeError(
f"Expected a datetime.datetime, got {type(value).__name__!r}"
)
if self.tz_aware and value.tzinfo is None:
raise ValueError(
f"Expected a timezone-aware datetime, got {value!r} instead."
)
[docs]
def serialize(self, value): # noqa: D102
return value.isoformat()
[docs]
def deserialize(self, value): # noqa: D102
if isinstance(value, str):
return datetime.datetime.fromisoformat(value)
return value
[docs]
def from_string(self, s): # noqa: D102
try:
return datetime.datetime.fromisoformat(s)
except ValueError as err:
raise ValueError(
f"Cannot parse {s!r} as ISO datetime (env var {self.env!r})"
) from err
[docs]
def to_string(self, value): # noqa: D102
return value.isoformat()
[docs]
def to_argparse_kwargs(self, name, prefix=""): # noqa: D102
kwargs = super().to_argparse_kwargs(name, prefix)
kwargs["metavar"] = "YYYY-MM-DDTHH:MM:SS"
return kwargs
#############################################################################
# Alias descriptor
#############################################################################
[docs]
class Alias:
"""A view field that delegates reads and writes to a dotpath on the
bound config.
Declare `Alias` attributes on a `ConfigView` subclass to define which
fields of the underlying config are exposed and under what names. The
argument is either a `FieldRef` obtained by class-level attribute access
on a `Config` class, or a plain dotpath string::
class CalibSummaryView(ConfigView):
psf_kernel = Alias(PSFFittingConfig.kernel_estimate) # preferred
threshold = Alias("detection.threshold") # also works
Parameters
----------
ref : `FieldRef` or `str`
Path to the target field on the bound config. Pass a `FieldRef`
(from class-level attribute access on a `Config`) to keep the path
refactorable via IDE rename tools. A plain dotpath string is
accepted for convenience.
"""
def __init__(self, ref):
self._path = ref._path if isinstance(ref, FieldRef) else ref
[docs]
def __set_name__(self, owner, name: str):
self.name = name
[docs]
def __get__(self, obj, objtype=None):
if obj is None:
return FieldRef(self.name, None)
return walk(obj._alias_root, self._path)
[docs]
def __set__(self, obj, value):
walk_set(obj._alias_root, self._path, value)
#############################################################################
# Mirror descriptor
#############################################################################
[docs]
class Mirror:
"""A config field that keeps multiple dotpaths in sync.
Declare a ``Mirror`` on a ``Config`` class to enforce that two or more
fields always hold the same value. A write fans out to every path; a
read asserts all paths agree and returns the shared value::
class SyncedConfig(Config, components=[CameraConfig, DetectorConfig]):
gain = Mirror(CameraConfig.gain, DetectorConfig.gain)
Parameters
----------
*refs : `FieldRef` or `str`
Dotpaths (relative to the config instance) that must stay in sync.
Pass `FieldRef` objects obtained from class-level attribute access on
a ``Config`` for refactorable, IDE-navigable paths.
"""
def __init__(self, *refs):
self._paths = [r._path if isinstance(r, FieldRef) else r for r in refs]
[docs]
def __set_name__(self, owner, name: str):
self.name = name
[docs]
def __get__(self, obj, objtype=None):
if obj is None:
return FieldRef(self.name, None)
values = [walk(obj, p) for p in self._paths]
if len(set(values)) > 1:
raise ValueError(
f"Mirror {self.name!r} paths disagree: "
f"{dict(zip(self._paths, values, strict=True))}"
)
return values[0]
[docs]
def __set__(self, obj, value):
for p in self._paths:
walk_set(obj, p, value)