from __future__ import annotations
import collections
import collections.abc
import dataclasses
import datetime
import enum
import inspect
import itertools
import json
import os
import pathlib
import re
import sys
from typing import (
Any,
Callable,
Dict,
Generic,
List,
Literal,
Sequence,
Set,
Tuple,
Type,
TypeVar,
Union,
cast,
)
from typing_extensions import TYPE_CHECKING, assert_never, get_args, get_origin
from .._typing_compat import is_typing_literal, is_typing_union
if TYPE_CHECKING:
from ._registry import ConstructorRegistry
from .. import _fmtlib as fmt
from .. import _resolver, _strings
from ..conf import _markers
from ._backtracking import parse_with_backtracking
[docs]
class UnsupportedTypeAnnotationError(Exception):
"""Exception raised when a type annotation is not supported."""
def __init__(self, message: tuple[fmt._Text, ...]):
self.message = message
super().__init__()
_TIMEDELTA_ISO8601 = re.compile(
r"^(?P<sign>[-+])?P"
r"(?:(?P<weeks>\d+(?:\.\d+)?)W)?"
r"(?:(?P<days>\d+(?:\.\d+)?)D)?"
r"(?:T"
r"(?:(?P<hours>\d+(?:\.\d+)?)H)?"
r"(?:(?P<minutes>\d+(?:\.\d+)?)M)?"
r"(?:(?P<seconds>\d+(?:\.\d+)?)S)?"
r")?$"
)
def _parse_timedelta(s: str) -> datetime.timedelta:
# Plain numeric value is treated as seconds. OverflowError covers inputs
# like "inf" or "1e400", which parse as floats but exceed timedelta's range.
try:
return datetime.timedelta(seconds=float(s))
except ValueError:
pass
except OverflowError:
raise ValueError(f"Invalid timedelta {s!r}: value out of range.")
match = _TIMEDELTA_ISO8601.match(s)
if match is None:
raise ValueError(
f"Invalid timedelta {s!r}: expected an ISO 8601 duration "
"(e.g., 'P1DT2H30M', 'PT45.5S') or a number of seconds."
)
parts = match.groupdict()
sign = -1 if parts.pop("sign") == "-" else 1
kwargs = {k: float(v) for k, v in parts.items() if v is not None}
if not kwargs:
raise ValueError(
f"Invalid timedelta {s!r}: ISO 8601 duration must include at least "
"one component (e.g., 'PT0S')."
)
try:
return sign * datetime.timedelta(**kwargs)
except OverflowError:
raise ValueError(f"Invalid timedelta {s!r}: value out of range.")
def _format_timedelta(td: datetime.timedelta) -> str:
if td == datetime.timedelta(0):
return "PT0S"
sign = ""
if td < datetime.timedelta(0):
sign = "-"
td = -td
days = td.days
hours, rem = divmod(td.seconds, 3600)
minutes, seconds = divmod(rem, 60)
microseconds = td.microseconds
out = sign + "P"
if days:
out += f"{days}D"
if hours or minutes or seconds or microseconds:
out += "T"
if hours:
out += f"{hours}H"
if minutes:
out += f"{minutes}M"
if seconds or microseconds:
if microseconds:
fractional = f"{seconds}.{microseconds:06d}".rstrip("0").rstrip(".")
out += f"{fractional}S"
else:
out += f"{seconds}S"
return out
T = TypeVar("T")
[docs]
@dataclasses.dataclass(frozen=True)
class PrimitiveTypeInfo:
"""Information used to generate constructors for primitive types."""
type: Type
"""Annotated field type. Forward references, aliases, and type
variables/parameters will have been resolved and runtime annotations
(typing.Annotated) will have been stripped."""
type_origin: Type | None
"""The output of get_origin() on the static type."""
markers: set[_markers.Marker]
"""Set of tyro markers used to configure this field."""
_primitive_spec: PrimitiveConstructorSpec | None
"""Primitive constructor spec that was scraped from runtime annotations."""
[docs]
@staticmethod
def make(
raw_annotation: Type | Callable,
parent_markers: set[_markers.Marker],
exclude_markers: set[_markers.Marker] | None = None,
) -> PrimitiveTypeInfo:
_, primitive_specs = _resolver.unwrap_annotated(
raw_annotation, search_type=PrimitiveConstructorSpec
)
primitive_spec = primitive_specs[0] if len(primitive_specs) > 0 else None
typ, extra_markers = _resolver.unwrap_annotated(
raw_annotation, search_type=_markers._Marker
)
markers = parent_markers | set(extra_markers)
if exclude_markers is not None:
markers = markers - exclude_markers
return PrimitiveTypeInfo(
type=cast(Type, typ),
type_origin=get_origin(typ),
markers=markers,
_primitive_spec=primitive_spec,
)
[docs]
@dataclasses.dataclass(frozen=True)
class PrimitiveConstructorSpec(Generic[T]):
"""Specification for constructing a primitive type from a string.
There are two ways to use this class:
First, we can include it in a type signature via :class:`typing.Annotated`.
This is the simplest for making local modifications to parsing behavior for
individual fields.
Alternatively, it can be returned by a rule in a :class:`ConstructorRegistry`.
"""
nargs: int | tuple[int, ...] | Literal["*"]
"""Number of arguments required to construct an instance. If nargs is "*", then
the number of arguments is variable. If nargs is a tuple, it represents multiple
valid fixed argument counts (used for unions with different fixed nargs)."""
metavar: str
"""Metavar to display in help messages."""
instance_from_str: Callable[[list[str]], T]
"""Given a list of string arguments, construct an instance of the type. The
length of the list will match the value of nargs."""
is_instance: Callable[[Any], bool | Literal["~"]]
"""Given an object instance, does it match this primitive type? This is
used for specific help messages when both a union type is present and a
default is provided.
Can return "~" to signify that an instance is a "fuzzy" match, and should
only be used if there are no other matches. This is used for numeric tower
support.
"""
str_from_instance: Callable[[T], list[str]]
"""Convert an instance to a list of string arguments that would construct
the instance. This is used for help messages when a default is provided."""
choices: tuple[str, ...] | None = None
"""Finite set of choices for arguments."""
_action: Literal["append"] | None = None
"""Internal action to use. Not part of the public API."""
def _compute_total_nargs(
specs: Sequence[PrimitiveConstructorSpec],
) -> int | tuple[int, ...] | Literal["*"]:
"""Compute all possible total argument counts for a sequence of specs.
When specs have tuple nargs (representing multiple valid argument counts),
this computes all possible sums.
Note: This uses itertools.product which is exponential in the number of specs,
but in practice we're only combining 2-3 specs (e.g., dict keys and values,
or tuple elements). Could be optimized in the future.
Args:
specs: Sequence of PrimitiveConstructorSpec objects.
Returns:
Total nargs value: either a single int, a tuple of ints, or "*".
"""
nargs_options_per_spec: list[tuple[int, ...]] = []
is_variable_nargs = False
for spec in specs:
if spec.nargs == "*":
is_variable_nargs = True
elif isinstance(spec.nargs, int):
nargs_options_per_spec.append((spec.nargs,))
elif isinstance(spec.nargs, tuple):
nargs_options_per_spec.append(tuple(sorted(spec.nargs)))
else:
assert_never(spec.nargs)
if is_variable_nargs:
# If any spec has nargs='*', the total nargs is variable.
return "*"
else:
possible_totals = {
sum(combo) for combo in itertools.product(*nargs_options_per_spec)
}
nargs = tuple(sorted(possible_totals))
if len(nargs) == 1:
return nargs[0]
return nargs
[docs]
def apply_default_primitive_rules(registry: ConstructorRegistry) -> None:
"""Apply default rules to the registry."""
from ._registry import ConstructorRegistry
@registry.primitive_rule
def any_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
if type_info.type is not Any:
return None
return UnsupportedTypeAnnotationError(
(fmt.text("`Any` is not a parsable type."),)
)
# HACK (json.loads): this is for code that uses
# `tyro.conf.arg(constructor=json.loads)`. We're going to deprecate this
# syntax (the constructor= argument in tyro.conf.arg), but there is code
# that lives in the wild that relies on the behavior so we'll do our best
# not to break it.
vanilla_types = (int, str, float, complex, bytes, bytearray, json.loads)
@registry.primitive_rule
def basics_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if type_info.type not in vanilla_types:
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar=type_info.type.__name__.upper(),
instance_from_str=lambda args: (
# `bytes`/`bytearray` need an explicit encoding; the others
# accept the string directly.
type_info.type(args[0], "ascii")
if type_info.type in (bytes, bytearray)
else type_info.type(args[0])
),
# issubclass(type(x), y) here is preferable over isinstance(x, y)
# due to quirks in the numeric tower.
is_instance=lambda x: _resolver.isinstance_with_fuzzy_numeric_tower(
x, type_info.type
),
str_from_instance=lambda instance: [str(instance)],
)
if "torch" in sys.modules.keys():
import torch
@registry.primitive_rule
def torch_device_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | None:
if type_info.type is not torch.device: # pyright: ignore[reportPrivateImportUsage]
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar=type_info.type.__name__.upper(),
instance_from_str=lambda args: torch.device(args[0]), # pyright: ignore[reportPrivateImportUsage]
is_instance=lambda x: isinstance(x, type_info.type),
str_from_instance=lambda instance: [str(instance)],
)
@registry.primitive_rule
def bool_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if type_info.type is not bool:
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar="{True,False}",
instance_from_str=lambda args: args[0] == "True",
choices=("True", "False"),
is_instance=lambda x: isinstance(x, bool),
str_from_instance=lambda instance: ["True" if instance else "False"],
)
@registry.primitive_rule
def nonetype_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if type_info.type is not type(None):
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar="{None}",
choices=("None",),
instance_from_str=lambda args: None,
is_instance=lambda x: x is None,
str_from_instance=lambda instance: ["None"],
)
@registry.primitive_rule
def path_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
try:
if not (
type_info.type in (os.PathLike, pathlib.Path)
or issubclass(type_info.type, pathlib.PurePath)
):
return None
except TypeError:
# `issubclass()` failed.
return None
def path_from_str(args: List[str]) -> Any:
typ = type_info.type
# Pure path types (PurePath/PurePosixPath/PureWindowsPath) are
# instantiable on any OS and preserve their flavour, so honor the
# annotated type. Concrete `Path`/`os.PathLike` fall back to
# `pathlib.Path`, which resolves to the correct OS-specific class
# (and avoids NotImplementedError when instantiating the
# non-native concrete `PosixPath`/`WindowsPath`).
if (
isinstance(typ, type)
and issubclass(typ, pathlib.PurePath)
and not issubclass(typ, pathlib.Path)
):
return typ(args[0])
return pathlib.Path(args[0])
return PrimitiveConstructorSpec(
nargs=1,
metavar=type_info.type.__name__.upper(),
instance_from_str=path_from_str,
is_instance=lambda x: hasattr(x, "__fspath__"),
str_from_instance=lambda instance: [str(instance)],
)
@registry.primitive_rule
def upath_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
# Check if upath is imported.
if "upath" not in sys.modules.keys(): # pragma: no cover
return None
try:
import upath
UPath = upath.UPath
except (ImportError, AttributeError):
# Needed for the mock import test in
# test_missing_optional_packages.py to pass.
return None
try:
if not (type_info.type is UPath or (issubclass(type_info.type, UPath))):
return None
except TypeError:
# `issubclass()` failed.
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar=type_info.type.__name__.upper(),
instance_from_str=lambda args: type_info.type(args[0]),
is_instance=lambda x: isinstance(x, UPath),
str_from_instance=lambda instance: [str(instance)],
)
@registry.primitive_rule
def enum_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if not (
inspect.isclass(type_info.type) and issubclass(type_info.type, enum.Enum)
):
return None
cast_type = cast(Type[enum.Enum], type_info.type)
if _markers.EnumChoicesFromValues in type_info.markers:
choices = tuple(str(m.value) for m in cast_type)
else:
choices = tuple(type_info.type.__members__.keys())
return PrimitiveConstructorSpec(
nargs=1,
metavar="{" + ",".join(choices) + "}",
instance_from_str=lambda args: (
next(
iter(member for member in cast_type if str(member.value) == args[0])
)
if _markers.EnumChoicesFromValues in type_info.markers
else cast_type[args[0]]
),
is_instance=lambda x: isinstance(x, cast_type),
str_from_instance=lambda instance: [
str(instance.value)
if _markers.EnumChoicesFromValues in type_info.markers
else instance.name
],
choices=choices,
)
@registry.primitive_rule
def datetime_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if type_info.type not in (datetime.datetime, datetime.date, datetime.time):
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar={
datetime.datetime: "YYYY-MM-DD[THH:MM:[SS[…]]]",
datetime.date: "YYYY-MM-DD",
datetime.time: "HH:MM[:SS[…]]",
}[type_info.type],
instance_from_str=lambda args: cast(
Union[
Type[datetime.datetime], Type[datetime.date], Type[datetime.time]
],
type_info.type,
).fromisoformat(args[0]),
is_instance=lambda x: isinstance(x, type_info.type),
str_from_instance=lambda instance: [instance.isoformat()],
)
@registry.primitive_rule
def timedelta_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if type_info.type is not datetime.timedelta:
return None
return PrimitiveConstructorSpec(
nargs=1,
metavar="SECONDS|P[nD][T[nH][nM][nS]]",
instance_from_str=lambda args: _parse_timedelta(args[0]),
is_instance=lambda x: isinstance(x, datetime.timedelta),
str_from_instance=lambda instance: [_format_timedelta(instance)],
)
@registry.primitive_rule
def vague_container_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
if type_info.type not in (
dict,
Dict,
tuple,
Tuple,
list,
List,
collections.abc.Sequence,
Sequence,
set,
Set,
):
return None
typ = type_info.type
if typ in (dict, Dict):
typ = Dict[str, str]
elif typ in (tuple, Tuple):
typ = Tuple[str, ...] # type: ignore
elif typ in (list, List, collections.abc.Sequence, Sequence):
typ = List[str]
elif typ in (set, Set):
typ = Set[str]
return ConstructorRegistry.get_primitive_spec(
PrimitiveTypeInfo.make(
typ,
parent_markers=type_info.markers,
)
)
@registry.primitive_rule
def sequence_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
if type_info.type_origin not in (
collections.abc.Sequence,
collections.abc.MutableSequence,
collections.abc.Set,
collections.abc.MutableSet,
frozenset,
list,
set,
collections.deque,
tuple,
):
return None
container_type = type_info.type_origin
assert container_type is not None
# Map abstract base classes to concrete constructible types. The
# immutable `collections.abc.Set` maps to `frozenset`; the mutable abcs
# map to their natural mutable concrete type.
if container_type in (
collections.abc.Sequence,
collections.abc.MutableSequence,
):
container_type = list
elif container_type is collections.abc.Set:
container_type = frozenset
elif container_type is collections.abc.MutableSet:
container_type = set
args = get_args(type_info.type)
if container_type is tuple:
assert len(args) == 2
(contained_type, ell) = args
assert ell == Ellipsis
elif len(args) == 1:
(contained_type,) = args
else:
contained_type = Any
inner_spec = ConstructorRegistry.get_primitive_spec(
PrimitiveTypeInfo.make(
raw_annotation=contained_type,
parent_markers=type_info.markers - {_markers.UseAppendAction},
)
)
if isinstance(inner_spec, UnsupportedTypeAnnotationError):
return UnsupportedTypeAnnotationError(
(
fmt.text(
"Could not create sequence primitive spec from ",
fmt.text["cyan"](f"`{type_info.type}`"),
" due to unsupported inner type",
),
*inner_spec.message,
)
)
# We can now handle nargs='*' with backtracking, so no need to reject it.
def instance_from_str(args: list[str]) -> Any:
result = parse_with_backtracking(
args=args,
specs=(inner_spec,),
is_repeating=True,
)
if result is None:
raise ValueError(f"Could not find valid parse for arguments: {args}")
out = result
assert container_type is not None
return cast(Callable, container_type)(out)
def str_from_instance(instance: Sequence) -> list[str]:
out = []
for i in instance:
out.extend(inner_spec.str_from_instance(i))
return out
if _markers.UseAppendAction in type_info.markers:
return PrimitiveConstructorSpec(
nargs=inner_spec.nargs,
metavar=inner_spec.metavar,
instance_from_str=inner_spec.instance_from_str,
is_instance=lambda x: (
isinstance(x, container_type)
and all(inner_spec.is_instance(i) for i in x)
),
str_from_instance=str_from_instance,
choices=inner_spec.choices,
_action="append",
)
else:
return PrimitiveConstructorSpec(
nargs="*",
metavar=_strings.multi_metavar_from_single(inner_spec.metavar),
instance_from_str=instance_from_str,
is_instance=lambda x: (
isinstance(x, container_type)
and all(inner_spec.is_instance(i) for i in x)
),
str_from_instance=str_from_instance,
choices=inner_spec.choices,
)
@registry.primitive_rule
def tuple_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
if type_info.type_origin is not tuple:
return None
types = get_args(type_info.type)
# The empty tuple type `Tuple[()]` is represented inconsistently across
# Python versions: `get_args` returns `()` on 3.11+ but `((),)` on
# 3.8-3.10. Normalize the latter to "no inner types" so both produce a
# spec that consumes zero arguments.
if types == ((),):
types = () # pragma: no cover (reachable only on Python 3.8-3.10)
typeset = set(types) # Sets are unordered.
typeset_no_ellipsis = typeset - {Ellipsis} # type: ignore
if typeset_no_ellipsis != typeset:
# Ellipsis: variable argument counts. When an ellipsis is used, tuples must
# contain only one type.
assert len(typeset_no_ellipsis) == 1
return sequence_rule(type_info)
inner_specs: list[PrimitiveConstructorSpec] = []
for contained_type in types:
spec = ConstructorRegistry.get_primitive_spec(
PrimitiveTypeInfo.make(contained_type, type_info.markers)
)
if isinstance(spec, UnsupportedTypeAnnotationError):
return UnsupportedTypeAnnotationError(
(
fmt.text(
"Could not create tuple primitive spec from ",
fmt.text["cyan"](f"`{type_info.type}`"),
" due to unsupported inner type",
),
*spec.message,
)
)
inner_specs.append(spec)
def instance_from_str(args: list[str]) -> tuple:
# Use backtracking for all cases (both fixed and variable nargs).
# Complexity is bad, O(k^n), where k is the max number of nargs.
# options and n is the number of tuple elements. We could revisit,
# but in practice k and n should both be small.
result = parse_with_backtracking(
args=args,
specs=tuple(inner_specs),
is_repeating=False,
)
if result is None:
raise ValueError(
f"Could not parse arguments {args} into tuple type {type_info.type}"
)
return tuple(result)
def str_from_instance(instance: tuple) -> list[str]:
out = []
for member, spec in zip(instance, inner_specs):
out.extend(spec.str_from_instance(member))
return out
# Compute all possible total argument counts.
nargs = _compute_total_nargs(inner_specs)
return PrimitiveConstructorSpec(
nargs=nargs,
metavar=" ".join(spec.metavar for spec in inner_specs),
instance_from_str=instance_from_str,
str_from_instance=str_from_instance,
is_instance=lambda x: (
isinstance(x, tuple)
and len(x) == len(inner_specs)
and all(
spec.is_instance(member) for member, spec in zip(x, inner_specs)
)
),
)
@registry.primitive_rule
def dict_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
# Map the (possibly abstract or subclass) mapping origin to a concrete
# factory that turns a plain parsed `dict` into the desired type.
#
# Notes:
# - `collections.abc.Mapping` is immutable in spirit, but there is no
# immutable concrete mapping in the stdlib, so (matching prior
# behavior) we construct a plain `dict`.
# - `defaultdict` is constructed as `defaultdict(None, parsed)`. We
# cannot infer a sensible `default_factory` from the annotation, so we
# leave it as `None`. The result is still a real `defaultdict`
# instance and round-trips through `str_from_instance`; it simply
# behaves like a plain dict on missing-key access (raising KeyError).
mapping_origin = type_info.type_origin
mapping_factories: dict[Any, Callable[[dict], Any]] = {
dict: dict,
collections.abc.Mapping: dict,
collections.abc.MutableMapping: dict,
collections.OrderedDict: collections.OrderedDict,
collections.Counter: collections.Counter,
collections.defaultdict: lambda parsed: collections.defaultdict(
None, parsed
),
}
if mapping_origin not in mapping_factories:
return None
container_factory = mapping_factories[mapping_origin]
type_args = get_args(type_info.type)
if mapping_origin is collections.Counter:
# `Counter[K]` has a single type argument: the key type. The values
# are implicitly integer counts.
if len(type_args) != 1:
return None
(key_type,) = type_args
val_type: Any = int
elif len(type_args) == 2:
key_type, val_type = type_args
else:
return None
key_spec = ConstructorRegistry.get_primitive_spec(
PrimitiveTypeInfo.make(
raw_annotation=key_type,
parent_markers=type_info.markers,
)
)
val_spec = ConstructorRegistry.get_primitive_spec(
PrimitiveTypeInfo.make(
raw_annotation=val_type,
parent_markers=type_info.markers - {_markers.UseAppendAction},
)
)
if isinstance(key_spec, UnsupportedTypeAnnotationError):
return UnsupportedTypeAnnotationError(
(
fmt.text(
"Could not create dict primitive spec from ",
fmt.text["cyan"](f"`{type_info.type}`"),
" due to unsupported key type",
),
*key_spec.message,
)
)
if isinstance(val_spec, UnsupportedTypeAnnotationError):
return UnsupportedTypeAnnotationError(
(
fmt.text(
"Could not create dict primitive spec from ",
fmt.text["cyan"](f"`{type_info.type}`"),
" due to unsupported value type",
),
*val_spec.message,
)
)
pair_metavar = f"{key_spec.metavar} {val_spec.metavar}"
def instance_from_str(args: list[str]) -> dict:
out = {}
# For UseAppendAction, we need to determine if we're parsing:
# 1. A single key-value pair (when nargs is fixed).
# 2. Multiple key-value pairs (when nargs is '*').
if _markers.UseAppendAction in type_info.markers:
# Check if we have a fixed number of args for a single pair.
if isinstance(key_spec.nargs, int) and isinstance(val_spec.nargs, int):
# Fixed size: parse single key-value pair.
is_repeating = False
else:
# Variable size: parse multiple pairs.
is_repeating = True
else:
# Without UseAppendAction, always parse multiple pairs.
is_repeating = True
parsed = parse_with_backtracking(
args, (key_spec, val_spec), is_repeating=is_repeating
)
if parsed is None:
raise ValueError("Failed to parse key-value pairs!")
# When is_repeating=True, parse_with_backtracking alternates between
# key_spec and val_spec, so parsed contains [key1, val1, key2, val2, ...].
for i in range(0, len(parsed), 2):
out[parsed[i]] = parsed[i + 1]
return container_factory(out)
def str_from_instance(instance: dict) -> list[str]:
# TODO: this may be strange right now for the append action.
out: list[str] = []
for key, value in instance.items():
out.extend(key_spec.str_from_instance(key))
out.extend(val_spec.str_from_instance(value))
return out
if _markers.UseAppendAction in type_info.markers:
# Compute all possible total argument counts for dict key-value pairs.
nargs = _compute_total_nargs([key_spec, val_spec])
return PrimitiveConstructorSpec(
nargs=nargs,
metavar=pair_metavar,
instance_from_str=instance_from_str,
is_instance=lambda x: (
isinstance(x, dict)
and all(
key_spec.is_instance(k) and val_spec.is_instance(v)
for k, v in x.items()
)
),
str_from_instance=str_from_instance,
_action="append",
)
else:
return PrimitiveConstructorSpec(
nargs="*",
metavar=_strings.multi_metavar_from_single(pair_metavar),
instance_from_str=instance_from_str,
is_instance=lambda x: (
isinstance(x, dict)
and all(
key_spec.is_instance(k) and val_spec.is_instance(v)
for k, v in x.items()
)
),
str_from_instance=str_from_instance,
)
@registry.primitive_rule
def literal_rule(type_info: PrimitiveTypeInfo) -> PrimitiveConstructorSpec | None:
if not is_typing_literal(type_info.type_origin):
return None
choices = get_args(type_info.type)
str_choices = tuple(
(
(
x.value
if _markers.EnumChoicesFromValues in type_info.markers
else x.name
)
if isinstance(x, enum.Enum)
else str(x)
)
for x in choices
)
return PrimitiveConstructorSpec(
nargs=1,
metavar="{" + ",".join(str_choices) + "}",
instance_from_str=lambda args: choices[str_choices.index(args[0])],
is_instance=lambda x: x in choices,
str_from_instance=lambda instance: [str(instance)],
choices=str_choices,
)
@registry.primitive_rule
def union_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
if not is_typing_union(type_info.type_origin):
return None
options = list(get_args(type_info.type))
if type(None) in options:
# Move `None` types to the beginning.
# If we have `Optional[str]`, we want this to be parsed as
# `Union[NoneType, str]`.
options.remove(type(None))
options.insert(0, type(None))
# General unions, eg Union[int, bool]. We'll try to convert these from left to
# right.
option_specs: dict[Type[object], PrimitiveConstructorSpec] = {}
choices: tuple[str, ...] | None = ()
nargs: int | tuple[int, ...] | Literal["*"] = 1
first = True
all_fixed_nargs = True
nargs_set: set[int] = set()
metavar_parts: list[str] = []
for t in options:
option_type_info = PrimitiveTypeInfo.make(
raw_annotation=t,
parent_markers=type_info.markers,
)
# If the argument is not suppressed, we can add the ability to
# suppress individual options.
if (
_markers.Suppress not in type_info.markers
and _markers.Suppress in option_type_info.markers
):
continue
option_spec = ConstructorRegistry.get_primitive_spec(option_type_info)
if isinstance(option_spec, UnsupportedTypeAnnotationError):
return UnsupportedTypeAnnotationError(
(
fmt.text(
"Could not create union primitive spec from ",
fmt.text["cyan"](f"`{type_info.type}`"),
" due to unsupported option type",
),
*option_spec.message,
)
)
if option_spec.choices is None:
choices = None
elif choices is not None:
choices = choices + option_spec.choices
option_specs[t] = option_spec
if not (t is type(None) and _markers.DisallowNone in type_info.markers):
metavar_parts.append(option_spec.metavar)
# Track if all options have fixed nargs. We include NoneType here
# (its nargs is always 1) so that unions like
# `Optional[Tuple[int, int]]` produce a nargs that accepts both the
# single-token `None` and the multi-token tuple. Without this, the
# computed nargs would be the tuple's exact count and the `None`
# token could never be supplied.
if isinstance(option_spec.nargs, int):
nargs_set.add(option_spec.nargs)
elif t is not type(None):
all_fixed_nargs = False
if t is not type(None):
# Enforce that `nargs` is the same for all child types, except for
# NoneType.
if first:
nargs = option_spec.nargs
first = False
elif nargs != option_spec.nargs:
# If all options have fixed nargs (even if different), collect them.
if all_fixed_nargs and isinstance(option_spec.nargs, int):
continue # We'll handle this after the loop.
else:
# Just be as general as possible if we see inconsistencies.
nargs = "*"
# If we have multiple fixed nargs values, use a tuple.
if all_fixed_nargs and len(nargs_set) > 1:
nargs = tuple(sorted(nargs_set))
metavar: str
metavar = _strings.join_union_metavars(metavar_parts)
def union_instantiator(strings: list[str]) -> Any:
errors = []
for i, (t, option_spec) in enumerate(option_specs.items()):
# Check if NoneType is disallowed.
if t is type(None) and _markers.DisallowNone in type_info.markers:
continue
# Check choices.
if option_spec.choices is not None and any(
x not in option_spec.choices for x in strings
):
errors.append(
f"{options[i]}: {strings} does not match choices {option_spec.choices}"
)
continue
# Try passing input into instantiator.
nargs_match = False
if isinstance(option_spec.nargs, int):
nargs_match = len(strings) == option_spec.nargs
elif isinstance(option_spec.nargs, tuple):
nargs_match = len(strings) in option_spec.nargs
elif option_spec.nargs == "*":
nargs_match = True
if nargs_match:
try:
return option_spec.instance_from_str(strings)
except ValueError as e:
# Failed, try next instantiator.
errors.append(f"{options[i]}: {e.args[0]}")
else:
errors.append(
f"{options[i]}: input length {len(strings)} did not match expected"
f" argument count {option_spec.nargs} of `[bold]{option_spec.metavar}[/bold]`"
)
raise ValueError(
f"no type matching {metavar} could be instantiated from"
f" {strings}.\n\nGot errors: \n- " + "\n- ".join(errors)
)
def str_from_instance(instance: Any) -> list[str]:
fuzzy_match = None
for option_spec in option_specs.values():
is_instance = option_spec.is_instance(instance)
if is_instance is True:
return option_spec.str_from_instance(instance)
elif is_instance == "~":
fuzzy_match = option_spec
# If we get here, we have a fuzzy match.
if fuzzy_match is not None:
return fuzzy_match.str_from_instance(instance)
assert False, (
f"could not match default value {instance} with any types in union {options}"
)
return PrimitiveConstructorSpec(
nargs=nargs,
metavar=metavar,
instance_from_str=union_instantiator,
is_instance=lambda x: any(
spec.is_instance(x) for spec in option_specs.values()
),
str_from_instance=str_from_instance,
choices=None if choices is None else tuple(set(choices)),
)
@registry.primitive_rule
def python_syntax_collections_rule(
type_info: PrimitiveTypeInfo,
) -> PrimitiveConstructorSpec | UnsupportedTypeAnnotationError | None:
"""Handle collections with Python literal syntax when UsePythonSyntaxForLiteralCollections marker is present."""
# Check if the marker is present.
if _markers.UsePythonSyntaxForLiteralCollections not in type_info.markers:
return None
# Check if the type is a collection type.
if type_info.type_origin not in (list, tuple, set, dict, frozenset):
return None
# Return None if no type arguments.
type_args = get_args(type_info.type)
if len(type_args) == 0:
return None
# Validate that all innermost types are compatible with ast.literal_eval.
def is_literal_eval_compatible(typ: Any) -> bool:
"""Check if a type is compatible with ast.literal_eval().
ast.literal_eval() only supports: str, bytes, int, float, complex, bool, None,
and the collection types list, tuple, dict, set. typing.Literal types are also
supported. Nested structures of these types are supported.
"""
from typing import Literal as LiteralType
# Ellipsis is used in variable-length tuples like tuple[int, ...].
if typ is Ellipsis:
return True
origin = get_origin(typ)
# Handle Literal types - check if all values are literal-eval compatible.
if origin is LiteralType:
# Literal values themselves must be literal-eval compatible types.
return all(
type(arg) in (str, bytes, int, float, complex, bool, type(None))
for arg in get_args(typ)
)
if origin is not None:
# Recurse into generic types.
return all(is_literal_eval_compatible(arg) for arg in get_args(typ))
# Whitelist only types that ast.literal_eval() actually supports.
# This excludes built-in types like frozenset, range, slice, etc.
return typ in (
str,
bytes,
int,
float,
complex,
bool,
type(None),
list,
tuple,
dict,
set,
)
# Return None if any types are incompatible - let other rules handle it.
if not all(is_literal_eval_compatible(arg) for arg in type_args):
return None
def instance_from_str(args: list[str]) -> Any:
import ast
try:
# Use ast.literal_eval() for safe parsing of Python literals.
# Only supports: strings, bytes, numbers, tuples, lists, dicts, sets,
# booleans, None, and nested structures of these types.
value = ast.literal_eval(args[0])
except (ValueError, SyntaxError) as e:
raise ValueError(
f"Could not parse '{args[0]}' as Python literal: {e}"
) from e
# Validate the type using typeguard.
if not _resolver.is_instance(type_info.type, value):
raise ValueError(
f"Value {value} (type {type(value)}) does not match expected type {type_info.type}"
)
return value
def str_from_instance(instance: Any) -> list[str]:
return [repr(instance)]
def is_instance_fn(x: Any) -> bool:
return _resolver.is_instance(type_info.type, x)
# Build a nice metavar showing the structure.
def get_metavar_for_type(typ: Any) -> str:
"""Recursively build metavar string for a type."""
from typing import Literal as LiteralType
origin = get_origin(typ)
# Handle Literal types - show the literal values.
if origin is LiteralType:
args = get_args(typ)
# Format as {val1,val2,val3}.
return "{" + ",".join(repr(arg) for arg in args) + "}"
if origin is None:
# Handle None type specially - show as "None" instead of "NONETYPE".
if typ is type(None):
return "None"
# Primitive type - just use uppercase name.
return typ.__name__.upper()
# Get type arguments.
args = get_args(typ)
if origin in (list, List):
elem = get_metavar_for_type(args[0])
return f"[{elem},...]"
elif origin in (tuple, Tuple):
if len(args) == 2 and args[1] is Ellipsis:
# Variable-length tuple.
elem = get_metavar_for_type(args[0])
return f"({elem},...)"
else:
# Fixed-length tuple.
elems = ",".join(get_metavar_for_type(arg) for arg in args)
return f"({elems})"
elif origin in (set, Set):
elem = get_metavar_for_type(args[0])
return f"{{{elem},...}}"
elif origin in (dict, Dict):
key = get_metavar_for_type(args[0])
val = get_metavar_for_type(args[1])
return f"{{{key}:{val},...}}"
else:
return str(origin).upper()
metavar = get_metavar_for_type(type_info.type)
return PrimitiveConstructorSpec(
nargs=1,
metavar=metavar,
instance_from_str=instance_from_str,
str_from_instance=str_from_instance,
is_instance=is_instance_fn,
)