Skip to content

pygx.typing

Runtime typing for symbolic objects: schemas, fields, and value specs.

typing

Runtime typing for symbolic objects: schemas, fields, and value specs.

Overview

To enable symbolic programmability on classes and functions, PyGX intercepts the assignment operation on the attributes of symbolic objects, to achieve two goals:

  • Enable automatic type checking, value validation, conversion and transformation on attribute values. See the Runtime typing section below for more details.

  • Allow symbolic attributes to be placeheld by markers, in order to represent abstract concepts such as a space of objects. E.g. hyper primitives like pygx.oneof mark a symbolic attribute to create a space of values for that attribute. See the Markers section below for more details.

Runtime typing

Symbolic objects are intended to be manipulated after creation. Without a runtime typing system, things can go wrong easily. For instance, an int attribute which was mistakenly modified at early program stages can be very difficult to debug at later stages. PyGX introduces a runtime type system that automatically validates symbolic objects upon creation and modification, minimizing boilerplated code for input validation, so the developer can focus on the main business logic.

Understanding schema

PyGX's runtime type system is based on the concept of Schema (class pygx.Schema), which defines what symbolic attributes are held by a symbolic type (e.g. a symbolic dict, a symbolic list or a symbolic class) and what values each attribute accepts. A Schema object consists of a list of Field (class pygx.Field), which define the acceptable keys (class pygx.KeySpec) and their values (class pygx.ValueSpec) for these types. A Schema object is usually created automatically and associated with a symbolic type upon its declaration, through dataclass-like field annotations on pygx.Object or decorators such as pygx.symbolize or pygx.functor. For example:

class A(pg.Object):
  x: int = 1
  y: float | None = None

print(A.__schema__)

@pg.symbolize([
    ('a', pg.typing.Int()),
    ('b', pg.typing.Float())
])
def foo(a, b):
  return a + b

print(foo.__schema__)

The first argument of pg.symbolize takes a list of field definitions, with each described by a tuple of 4 items:

(key specification, value specification, doc string, field metadata)

The key specification and value specification are required, while the doc string and the field metadata are optional. The key specification defines acceptable identifiers for this field, and the value specification defines the attribute's value type, its default value, validation rules. The doc string will serve as the description for the field, and the field metadata can be used for field-based code generation.

The following code snippet illustrates all supported KeySpec and ValueSpec subclasses and their usage with a manually created schema:

class A(pg.Object):
  x: int = 1

schema = pg.typing.create_schema([
    # Primitive types.
    ('a', pg.typing.Bool(default=True).noneable()),
    ('b', True),       # Equivalent to ('b', pg.typing.Bool(default=True)).
    ('c', pg.typing.Int()),
    ('d', 0),          # Equivalent to ('d', pg.typing.Int(default=0)).
    ('e', pg.typing.Int(min_value=0, max_value=10).noneable()),
    ('f', pg.typing.Float()),
    ('g', 1.0),        # Equivalent to ('g', pg.typing.Float(default=1.0)).
    ('h', pg.typing.Str()),
    ('i', 'foo'),      # Equivalent to ('i', pg.typing.Str(default='foo')).
    ('j', pg.typing.Str(regex='foo.*')),

    # Enum type.
    ('l', pg.typing.Enum(['foo', 'bar', 0, 1], 'foo')),

    # List type.
    ('m', pg.typing.List(pg.typing.Int(), size=2, default=[0, 0])),
    ('n', pg.typing.List(pg.typing.Dict([
        ('n1', pg.typing.List(pg.typing.Int())),
        ('n2', pg.typing.Str().noneable()),
    ]), min_size=1, max_size=10)),

    # Dict type.
    ('o', pg.typing.Dict([
        ('o1', pg.typing.Int()),
        ('o2', pg.typing.List(pg.typing.Dict([
            ('o21', 1),
            ('o22', 1.0),
        ]))),
        ('o3', pg.typing.Dict([
            # Use of regex key.
            (pg.typing.StrKey('n3.*'), pg.typing.Int()),
        ])),
    ])),

    # Tuple type.
    ('p', pg.typing.Tuple([pg.typing.Int(), pg.typing.Str()])),

    # Object type.
    ('q', pg.typing.Object(A, default=A())),

    # Type type.
    ('r', pg.typing.Type(int)),

    # Callable type.
    ('s', pg.typing.Callable([pg.typing.Int(), pg.typing.Int()],
                             kw=[('a', pg.typing.Str())])),

    # Functor type (same as Callable, but only for symbolic.Functor).
    ('t', pg.typing.Functor([pg.typing.Str()],
                            kw=[('a', pg.typing.Str())])),

    # Union type.
    ('u', pg.typing.Union([pg.typing.Int(), pg.typing.Str()], default=1)),

    # Any type.
    ('v', pg.typing.Any(default=1)),
])

From Python annotations

For most classes there is no need to hand-build value specs: standard Python type annotations on a pygx.Object subclass convert to value specs automatically. The conversion rules:

  • int, str, bool, float map to pygx.typing.Int, pygx.typing.Str, pygx.typing.Bool, pygx.typing.Float.
  • X | None (or typing.Optional[X]) makes the field accept None; other unions (A | B) map to pygx.typing.Union.
  • typing.Literal[...] maps to pygx.typing.Enum.
  • list[E] / dict[str, V] map to pygx.typing.List / pygx.typing.Dict specs holding plain list / dict values, while pg.List[E] / pg.Dict[str, V] annotations produce the same specs with symbolic storage: assigned raw containers are converted into symbolic pg.List / pg.Dict instances. A trailing typing.Literal[False] type argument (e.g. pg.List[E, typing.Literal[False]]) sets auto_wrap=False: the annotated container itself is still symbolic, but raw dict / list values landing in member positions the spec leaves unconstrained (e.g. Any-typed elements) are kept verbatim instead of being wrapped into their symbolic forms.
  • A typing.TypedDict subclass maps to a pygx.typing.Dict spec mirroring its keys; extra keys are rejected unless the TypedDict declares PEP 728 extra_items=....
  • PEP 695 type aliases (type IntList = list[int]), including subscripted generic aliases (Pair[int]), expand to their aliased values.
  • Stdlib scalars such as datetime.datetime, uuid.UUID, decimal.Decimal and pathlib.Path map to dedicated specs (pygx.typing.Datetime, pygx.typing.Uuid, ...); any other class maps to pygx.typing.Object.

ValueSpec instances themselves are not accepted as annotations (static type checkers cannot read them). To attach an explicit value spec to a field, pass it via pg.field(value_spec=...) — either as the field's default or inside typing.Annotated:

import typing

class A(pg.Object):
  x: typing.Annotated[int, pg.field(value_spec=pg.typing.Int(min_value=0))]
  y: int = pg.field(default=1, value_spec=pg.typing.Int(max_value=10))

To validate a bare value against an annotation in one line, use pygx.validate.

Schema inheritance

Symbolic attributes can be inherited during subclassing. Accordingly, the schema that defines a symbolic class' attributes can be inherited too by its subclasses. The fields from the bases' schema will be carried over into the subclasses' schema, while the subclass can override, by redefining that field with the same key. The subclass cannot override its base classes' field with arbitrary value specs, it must be overriding non-frozen fields with more restrictive validation rules of the same type, or change their default values. See pygx.ValueSpec.extend for more details.

The code snippet below illustrates schema inheritance during subclassing:

import typing

class A(pg.Object):
  x: typing.Annotated[
      int, pg.field(value_spec=pg.typing.Int(min_value=1))]
  y: float

class B(A):
  # Further restrict inherited 'x' by specifying the max value, as well
  # as providing a default value.
  x: typing.Annotated[
      int, pg.field(value_spec=pg.typing.Int(max_value=5, default=2))]
  z: str = pg.field(value_spec=pg.typing.Str(default='foo').freeze())

assert list(B.__schema__.fields.keys()) == ['x', 'y', 'z']

class C(B):
  # Raises: 'z' is frozen in class B and cannot be extended further.
  z: str

Automatic type conversions

When a value is assigned to an attribute whose value specification does not match the input value type, a coercion takes place automatically if the target type knows how to accept the input. A type opts in by defining a __pg_accept__(value) classmethod that returns a converted instance (or NotImplemented to decline). For example:

class A:
  def __init__(self, s):
    self._str = s

  def __eq__(self, other):
    return isinstance(other, self.__class__) and self._str == other._str

  @classmethod
  def __pg_accept__(cls, value):
    return cls(value) if isinstance(value, str) else NotImplemented

assert pg.typing.Object(A).apply('abc') == A('abc')

For builtin targets that cannot host a classmethod, use a field-level pg.field(transform=...) instead. The universal int → float widening is intrinsic to pg.typing.Float.

Markers

Markers (formerly "symbolic placeholding") enable scenarios that require an abstract (non-material) representation of objects - for example - a search space or a partially bound object. Such objects with marked attributes can only be used symbolically, which means they can be symbolically queried or manipulated, but not yet materialized to execute the business logic. For example:

@pg.functor()
def foo(x, y):
  return x + y

f = foo(x=pg.oneof(range(5)), y=pg.floatv(0.0, 2.0))

# Error: `x` and `y` are not materialized, thus the functor cannot be
# executed.
# f()

# But we can symbolically manipulate it:
f.sym_rebind(x=1, y=2)

# Returns 3
assert f() == 3

Markers are built on the pygx.CustomTyping interface, a mechanism to extend the typing system horizontally without modifying the pygx.ValueSpec of existing classes' attributes: CustomTyping subclasses take over value validation and transformation from an attribute's original ValueSpec via the pygx.CustomTyping.custom_apply method.

Following is an example of using CustomTyping to extend the schema system — a Ratio object that can sit in any float-typed attribute:

class Ratio(pg.typing.CustomTyping):

  def __init__(self, numerator, denominator):
    self.numerator = numerator
    self.denominator = denominator

  def custom_apply(
      self, path, value_spec, allow_partial, child_transform=None):
    if isinstance(value_spec, pg.typing.Float):
      # Validate the ratio's value against the spec's constraints
      # (e.g. min_value/max_value).
      value_spec.apply(self.numerator / self.denominator)
      # Short-circuit `.apply` and return the object itself as the
      # final value.
      return (False, self)
    else:
      raise ValueError(
          f'Ratio can only be applied to float type. (path={path})')

class Foo(pg.Object):
  x: float

# Ratio is accepted for all symbolic attributes with a Float value spec.
f = Foo(x=Ratio(1, 2))
assert isinstance(f.x, Ratio)

MissingValue

Bases: Formattable, WireConvertible

Value placeholder for an unassigned attribute.

pg.MISSING_VALUE is the canonical instance: field machinery uses it to mark a declared field whose value has not been provided (see MaybePartial). All MissingValue instances compare equal (and share one hash), it is wire-convertible, and the class cannot be subclassed -- the symbolic core detects the sentinel with an exact-type check on its hot paths.

Argument dataclass

Argument(
    name: str, kind: Kind, value_spec: ValueSpec, description: str | None = None
)

Definition for a callable argument.

Kind

Bases: Enum

Argument kind.

from_parameter classmethod
from_parameter(parameter: Parameter) -> Kind

Returns Argument.Kind from inspect.Parameter.

Source code in pygx/typing/_callable_signature.py
@classmethod
def from_parameter(
    cls, parameter: inspect.Parameter
) -> 'Argument.Kind':
    """Returns Argument.Kind from inspect.Parameter."""
    # Dict lookup is faster than a chain of `==` comparisons and
    # has stable dispatch cost regardless of which kind it is.
    return _PARAM_KIND_TO_ARG_KIND[parameter.kind]

from_annotation classmethod

from_annotation(
    name: str,
    kind: Kind,
    annotation: Any = empty,
    auto_typing: bool = False,
    parent_module: ModuleType | None = None,
) -> Argument

Creates an argument from annotation.

Source code in pygx/typing/_callable_signature.py
@classmethod
def from_annotation(
    cls,
    name: str,
    kind: Kind,
    annotation: Any = inspect.Parameter.empty,
    auto_typing: bool = False,
    parent_module: types.ModuleType | None = None,
) -> 'Argument':
    """Creates an argument from annotation."""
    return cls(
        name,
        kind,
        class_schema.ValueSpec.from_annotation(
            annotation, auto_typing=auto_typing, parent_module=parent_module
        ),
    )

from_parameter classmethod

from_parameter(
    param: Parameter,
    description: str | None = None,
    auto_typing: bool = True,
    parent_module: ModuleType | None = None,
) -> Argument

Creates an argument from inspect.Parameter.

Source code in pygx/typing/_callable_signature.py
@classmethod
def from_parameter(
    cls,
    param: inspect.Parameter,
    description: str | None = None,
    auto_typing: bool = True,
    parent_module: types.ModuleType | None = None,
) -> 'Argument':
    """Creates an argument from inspect.Parameter."""
    default = param.default
    kind = param.kind
    value_spec = class_schema.ValueSpec.from_annotation(
        param.annotation,
        auto_typing=auto_typing,
        parent_module=parent_module,
    )
    if default is not inspect.Parameter.empty:
        value_spec.set_default(default)

    if kind is inspect.Parameter.VAR_POSITIONAL:
        value_spec = class_schema.ValueSpec.ListType(value_spec, default=[])
    elif kind is inspect.Parameter.VAR_KEYWORD:
        value_spec = class_schema.ValueSpec.DictType(value_spec)
    return cls(
        param.name,
        _PARAM_KIND_TO_ARG_KIND[kind],
        value_spec,
        description=description,
    )

to_field

to_field() -> Field

Converts current argument to a pg.typing.Field object.

Source code in pygx/typing/_callable_signature.py
def to_field(self) -> class_schema.Field:
    """Converts current argument to a pg.typing.Field object."""
    if self.kind == Argument.Kind.VAR_KEYWORD:
        key: ks.KeySpecBase = ks.StrKey()
        assert isinstance(self.value_spec, vs.Dict)
        varkw_schema = self.value_spec.schema
        assert (
            varkw_schema is not None
            and varkw_schema.dynamic_field is not None
        )
        value = varkw_schema.dynamic_field.value
    else:
        key = ks.ConstStrKey(self.name)
        value = self.value_spec
    return class_schema.Field(key, value, description=self.description)

CallableType

Bases: Enum

Enum for Callable type.

Signature

Signature(
    callable_type: CallableType,
    name: str,
    module_name: str | None,
    args: list[Argument] | None = None,
    kwonlyargs: list[Argument] | None = None,
    varargs: Argument | None = None,
    varkw: Argument | None = None,
    return_value: ValueSpec | None = None,
    qualname: str | None = None,
    description: str | None = None,
)

Bases: Formattable

A callable's signature, with a value spec attached to each argument.

Constructor.

Parameters:

Name Type Description Default
callable_type CallableType

Type of callable.

required
name str

Function name.

required
module_name str | None

Module name.

required
args list[Argument] | None

Specification for positional arguments.

None
kwonlyargs list[Argument] | None

Specification for keyword-only arguments.

None
varargs Argument | None

Specification for wildcard list argument, e.g. 'args' is the name for *args.

None
varkw Argument | None

Specification for wildcard keyword argument, e.g. 'kwargs' is the name for **kwargs.

None
return_value ValueSpec | None

Optional value spec for return value.

None
qualname str | None

Optional qualified name.

None
description str | None

Optional description of the signature.

None
Source code in pygx/typing/_callable_signature.py
def __init__(
    self,
    callable_type: CallableType,
    name: str,
    module_name: str | None,
    args: list[Argument] | None = None,
    kwonlyargs: list[Argument] | None = None,
    varargs: Argument | None = None,
    varkw: Argument | None = None,
    return_value: class_schema.ValueSpec | None = None,
    qualname: str | None = None,
    description: str | None = None,
):
    """Constructor.

    Args:
      callable_type: Type of callable.
      name: Function name.
      module_name: Module name.
      args: Specification for positional arguments.
      kwonlyargs: Specification for keyword-only arguments.
      varargs: Specification for wildcard list argument, e.g. 'args' is the
        name for `*args`.
      varkw: Specification for wildcard keyword argument, e.g. 'kwargs' is
        the name for `**kwargs`.
      return_value: Optional value spec for return value.
      qualname: Optional qualified name.
      description: Optional description of the signature.
    """
    args = args or []
    self.callable_type = callable_type
    self.name = name
    self.module_name = module_name
    self.args = args or []
    self.kwonlyargs = kwonlyargs or []
    self.varargs = varargs
    self.varkw = varkw
    self.return_value = return_value
    self.qualname = qualname or name
    self.description = description

named_args property

named_args: list[Argument]

Returns all named arguments according to their declaration order.

arg_names property

arg_names

Returns names of all arguments according to their declaration order.

id property

id: str

Returns ID of the function.

has_varargs property

has_varargs: bool

Returns whether wildcard positional argument is present.

has_varkw property

has_varkw: bool

Returns whether wildcard keyword argument is present.

has_wildcard_args property

has_wildcard_args: bool

Returns whether any wildcard arguments are present.

get_value_spec

get_value_spec(name: str) -> ValueSpec | None

Returns Value spec for an argument name.

Parameters:

Name Type Description Default
name str

Argument name.

required

Returns:

Type Description
ValueSpec | None

ValueSpec for the requested argument. If name is not found, value spec of

ValueSpec | None

wildcard keyword argument will be used. None will be returned if name

ValueSpec | None

does not exist in signature and wildcard keyword is not accepted.

Source code in pygx/typing/_callable_signature.py
def get_value_spec(self, name: str) -> class_schema.ValueSpec | None:
    """Returns Value spec for an argument name.

    Args:
      name: Argument name.

    Returns:
      ValueSpec for the requested argument. If name is not found, value spec of
      wildcard keyword argument will be used. None will be returned if name
      does not exist in signature and wildcard keyword is not accepted.
    """
    for arg in self.named_args:
        if arg.name == name:
            return arg.value_spec
    if self.varkw is not None:
        assert isinstance(self.varkw.value_spec, vs.Dict)
        varkw_schema = self.varkw.value_spec.schema
        assert (
            varkw_schema is not None
            and varkw_schema.dynamic_field is not None
        )
        return varkw_schema.dynamic_field.value
    return None

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format current object.

Source code in pygx/typing/_callable_signature.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format current object."""
    return formatting.kvlist_str(
        [
            ('', self.id, ''),
            ('args', self.args, []),
            ('kwonlyargs', self.kwonlyargs, []),
            ('returns', self.return_value, None),
            ('varargs', self.varargs, None),
            ('varkw', self.varkw, None),
            ('description', self.description, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

annotate

annotate(
    args: (
        dict[FieldKeyDef, FieldValueDef] | list[FieldDef | Field] | None
    ) = None,
    return_value: ValueSpec | Any | None = None,
) -> Signature

Annotate arguments with extra typing.

Source code in pygx/typing/_callable_signature.py
def annotate(
    self,
    args: (
        dict[class_schema.FieldKeyDef, class_schema.FieldValueDef]
        | list[class_schema.FieldDef | class_schema.Field]
        | None
    ) = None,
    return_value: class_schema.ValueSpec | Any | None = None,
) -> 'Signature':
    """Annotate arguments with extra typing."""
    if return_value is not None:
        return_value = class_schema.ValueSpec.from_annotation(
            return_value, auto_typing=True
        )
        if topology.MISSING_VALUE != return_value.default:
            raise ValueError(
                'return value spec should not have default value.'
            )
        self.return_value = return_value

    if not args:
        return self

    schema = class_schema.create_schema(args, allow_nonconst_keys=True)  # pylint: disable=redefined-outer-name

    arg_fields: dict[str, class_schema.Field] = dict()
    varargs_field = None
    kwarg_field = None
    existing_names = set(self.arg_names)
    extra_arg_names = []

    for arg_name, field in schema.fields.items():
        if isinstance(arg_name, ks.StrKey):
            if kwarg_field is not None:
                raise KeyError(
                    f'{self.id}: multiple StrKey found in override args.'
                )
            kwarg_field = field
        else:
            assert isinstance(arg_name, (str, ks.ConstStrKey))
            if self.varargs and self.varargs.name == arg_name:
                varargs_field = field

            elif self.varkw and self.varkw.name == arg_name:
                if kwarg_field is not None:
                    raise KeyError(
                        f'{self.id}: multiple StrKey found in '
                        f'symbolic arguments declaration.'
                    )
                kwarg_field = field
            elif arg_name not in existing_names:
                if self.has_varkw:
                    extra_arg_names.append(arg_name)
                else:
                    raise KeyError(
                        f'{self.id}: found extra symbolic argument {arg_name.text!r}.'
                    )
            arg_fields[arg_name.text] = field

    def update_arg(arg: Argument, field: class_schema.Field):
        """Updates an argument with override field."""
        if arg.value_spec.has_default and (
            not field.value.has_default
            # Loose the default as user may mark it as noneable.
            or field.value.default is None
        ):
            field.value.set_default(
                arg.value_spec.default, root_path=topology.KeyPath(arg.name)
            )
        if arg.value_spec.default != field.value.default:
            if field.value.is_noneable and not arg.value_spec.has_default:
                # Special handling noneable which always comes with a default.
                field.value.set_default(topology.MISSING_VALUE)
            elif not (
                # Special handling Dict type which always has default.
                isinstance(field.value, class_schema.ValueSpec.DictType)
                and not arg.value_spec.has_default
            ):
                raise ValueError(
                    f'The annotated default value ({field.default_value}) of '
                    f'symbolic argument {arg.name!r} is not equal to the default '
                    f'value ({arg.value_spec.default}) specified from the function '
                    'signature.'
                )
        arg.value_spec = field.value
        if field.description:
            arg.description = field.description

    # Named arguments.
    for arg in self.named_args:
        field = arg_fields.get(arg.name)
        if field is not None:
            update_arg(arg, field)

    # Add extra arguments.
    for arg_name in extra_arg_names:
        field = arg_fields.get(arg_name)
        assert field is not None
        self.kwonlyargs.append(
            Argument(
                name=arg_name.text,
                kind=Argument.Kind.KEYWORD_ONLY,
                value_spec=field.value,
                description=field.description,
            )
        )

    # Update varargs.
    if varargs_field is not None:
        assert self.varargs is not None
        if not isinstance(
            varargs_field.value, class_schema.ValueSpec.ListType
        ):
            raise ValueError(
                f'Variable positional argument {self.varargs.name!r} should have a '
                'value of `pg.typing.List` type. '
                f'Encountered: {varargs_field.value!r}.'
            )
        update_arg(self.varargs, varargs_field)

    # Update kwarg.
    if kwarg_field is not None:
        assert self.varkw is not None
        value_spec = class_schema.ValueSpec.DictType(kwarg_field.value)
        self.varkw.value_spec = value_spec
        if kwarg_field.description:
            self.varkw.description = kwarg_field.description

    return self

fields

fields(remove_self: bool = True, include_return: bool = False) -> list[Field]

Returns the fields of this signature.

Source code in pygx/typing/_callable_signature.py
def fields(
    self,
    remove_self: bool = True,
    include_return: bool = False,
) -> list[class_schema.Field]:
    """Returns the fields of this signature."""
    fields = [
        arg.to_field()
        for i, arg in enumerate(self.args)
        if not remove_self or i > 0 or arg.name != 'self'
    ]
    if self.varargs:
        fields.append(self.varargs.to_field())
    fields.extend([arg.to_field() for arg in self.kwonlyargs])
    if self.varkw:
        fields.append(self.varkw.to_field())
    if include_return and self.return_value:
        fields.append(
            class_schema.Field('return', self.return_value, 'Return value.')
        )
    return fields

to_schema

to_schema(remove_self: bool = True, include_return: bool = False) -> Schema

Returns the schema of this signature.

Source code in pygx/typing/_callable_signature.py
def to_schema(
    self,
    remove_self: bool = True,
    include_return: bool = False,
) -> class_schema.Schema:
    """Returns the schema of this signature."""
    init_arg_list = [arg.name for arg in self.args]
    if init_arg_list and init_arg_list[0] == 'self':
        init_arg_list.pop(0)

    if self.varargs:
        init_arg_list.append(f'*{self.varargs.name}')

    return class_schema.Schema(
        self.fields(remove_self=remove_self, include_return=include_return),
        name=f'{self.module_name}.{self.qualname}',
        description=self.description,
        allow_nonconst_keys=True,
        metadata=dict(
            init_arg_list=init_arg_list,
            varargs_name=self.varargs.name if self.varargs else None,
            varkw_name=self.varkw.name if self.varkw else None,
            returns=self.return_value,
        ),
    )

from_schema classmethod

from_schema(
    schema: Schema,
    module_name: str,
    name: str,
    qualname: str | None = None,
    is_method: bool = True,
    use_init_arg_list: bool = False,
) -> Signature

Creates a signature from a schema object.

Parameters:

Name Type Description Default
schema Schema

A pg.typing.Schema object associated with a pg.Object.

required
module_name str

Module name for the signature.

required
name str

Function or method name of the signature.

required
qualname str | None

Qualname of the signature.

None
is_method bool

If True, self will be added in the signature as the first argument.

True
use_init_arg_list bool

If True, honor the schema's init_arg_list metadata to preserve positional argument order (used by Functor.__call__ so the original function signature is kept). Default is False: pg.Object.__init__ is synthesized keyword-only and subclasses that need positional arguments must override __init__ explicitly.

False

Returns:

Type Description
Signature

A signature object from the schema.

Source code in pygx/typing/_callable_signature.py
@classmethod
def from_schema(
    cls,
    schema: class_schema.Schema,  # pylint: disable=redefined-outer-name
    module_name: str,
    name: str,
    qualname: str | None = None,
    is_method: bool = True,
    use_init_arg_list: bool = False,
) -> 'Signature':
    """Creates a signature from a schema object.

    Args:
      schema: A `pg.typing.Schema` object associated with a `pg.Object`.
      module_name: Module name for the signature.
      name: Function or method name of the signature.
      qualname: Qualname of the signature.
      is_method: If True, `self` will be added in the signature as the first
        argument.
      use_init_arg_list: If True, honor the schema's ``init_arg_list``
        metadata to preserve positional argument order (used by
        ``Functor.__call__`` so the original function signature is
        kept). Default is False: ``pg.Object.__init__`` is synthesized
        keyword-only and subclasses that need positional arguments
        must override ``__init__`` explicitly.

    Returns:
      A signature object from the schema.
    """
    args = []
    if is_method:
        args.append(
            Argument.from_annotation(
                'self', Argument.Kind.POSITIONAL_OR_KEYWORD
            )
        )

    kwonlyargs = []
    varargs = None
    varkw = None

    if use_init_arg_list:
        # Honor `init_arg_list` so callers (Functor.__call__) keep the
        # original Python positional signature.
        init_arg_list = schema.metadata.get('init_arg_list', ())
        if init_arg_list and init_arg_list[-1].startswith('*'):
            arg_names = init_arg_list[:-1]
            vararg_name = init_arg_list[-1][1:]
        else:
            arg_names = init_arg_list
            vararg_name = None

        schema_fields = schema._fields  # pylint: disable=protected-access
        allow_nonconst = schema._allow_nonconst_keys  # pylint: disable=protected-access

        def get_arg_spec(arg_name):
            field = schema_fields.get(arg_name)
            if field is None and allow_nonconst:
                field = schema.get_field(arg_name)
            if field is None:
                raise ValueError(
                    f'Argument {arg_name!r} is not a symbolic field.'
                )
            return field.value

        # A required field that follows a defaulted field is promoted to
        # kw-only (Python forbids `def f(x=1, y)`); promotion is sticky.
        promote_remaining = False
        for n in arg_names:
            spec = get_arg_spec(n)
            if (
                not promote_remaining
                and args
                and args[-1].value_spec.has_default
                and not spec.has_default
            ):
                promote_remaining = True
            if promote_remaining:
                kwonlyargs.append(
                    Argument(
                        name=n,
                        kind=Argument.Kind.KEYWORD_ONLY,
                        value_spec=spec,
                    )
                )
            else:
                args.append(
                    Argument(
                        name=n,
                        kind=Argument.Kind.POSITIONAL_OR_KEYWORD,
                        value_spec=spec,
                    )
                )
        if vararg_name:
            varargs = Argument(
                name=vararg_name,
                kind=Argument.Kind.VAR_POSITIONAL,
                value_spec=get_arg_spec(vararg_name),
            )  # pytype: disable=attribute-error

        existing_names = set(arg_names)
        if vararg_name:
            existing_names.add(vararg_name)
        for key, field in schema.fields.items():
            if (
                str(key) in existing_names
                or field.frozen
                or not field.enable_init
            ):
                continue
            if key.is_const:
                kwonlyargs.append(
                    Argument(
                        name=str(key),
                        kind=Argument.Kind.KEYWORD_ONLY,
                        value_spec=field.value,
                    )
                )
            else:
                # pytype: disable=not-instantiable
                # pytype: disable=wrong-arg-count
                varkw = Argument(
                    name=schema.metadata.get('varkw_name', None)
                    or 'kwargs',
                    kind=Argument.Kind.VAR_KEYWORD,
                    value_spec=class_schema.ValueSpec.DictType(field.value),
                )
                # pytype: enable=wrong-arg-count
                # pytype: enable=not-instantiable
    else:
        # `pg.Object` synthesizes a keyword-only `__init__`: every
        # schema field becomes a keyword-only argument.
        for key, field in schema.fields.items():
            if field.frozen or not field.enable_init:
                continue
            if key.is_const:
                kwonlyargs.append(
                    Argument(
                        name=str(key),
                        kind=Argument.Kind.KEYWORD_ONLY,
                        value_spec=field.value,
                    )
                )
            else:
                # pytype: disable=not-instantiable
                # pytype: disable=wrong-arg-count
                varkw = Argument(
                    name=schema.metadata.get('varkw_name', None)
                    or 'kwargs',
                    kind=Argument.Kind.VAR_KEYWORD,
                    value_spec=class_schema.ValueSpec.DictType(field.value),
                )
                # pytype: enable=wrong-arg-count
                # pytype: enable=not-instantiable

    return Signature(
        callable_type=CallableType.FUNCTION,
        name=name,
        module_name=module_name,
        qualname=qualname,
        description=schema.description,
        args=args,
        kwonlyargs=kwonlyargs,
        varargs=varargs,
        varkw=varkw,
        return_value=schema.metadata.get('returns', None),
    )

from_callable classmethod

from_callable(
    callable_object: Callable[..., Any],
    auto_typing: bool = False,
    auto_doc: bool = False,
) -> Signature

Creates Signature from a callable object.

Source code in pygx/typing/_callable_signature.py
@classmethod
def from_callable(
    cls,
    callable_object: Callable[..., Any],
    auto_typing: bool = False,
    auto_doc: bool = False,
) -> 'Signature':
    """Creates Signature from a callable object."""
    if not callable(callable_object):
        raise TypeError(f'{callable_object!r} is not callable.')

    if isinstance(callable_object, _functor.Functor):
        assert callable_object.__signature__ is not None
        return callable_object.__signature__

    func = callable_object
    docstr = None
    if inspect.isclass(func):
        # A class that already carries a pygx `Signature` — a functor
        # class (`@pg.functor` returns a class) or a symbolized class
        # wrapper — IS its own signature; `inspect.signature` would
        # choke on the non-stdlib `__signature__` attribute with an
        # opaque `TypeError: unexpected object ... in __signature__
        # attribute` (#783).
        own_sig = getattr(func, '__signature__', None)
        if isinstance(own_sig, Signature):
            return own_sig
        callable_type = CallableType.METHOD
        try:
            sig = inspect.signature(func)
        except ValueError:
            sig = inspect.signature(func.__init__)

        if auto_doc:
            description = None
            args_doc = {}
            if func.__doc__:
                cls_doc = _docstr.DocStr.parse(func.__doc__)
                description = cls_doc.short_description
                args_doc.update(cls_doc.args)

            if func.__init__.__doc__:
                init_doc = _docstr.DocStr.parse(func.__init__.__doc__)
                args_doc.update(init_doc.args)
            docstr = _docstr.DocStr(
                _docstr.DocStrStyle.GOOGLE,
                short_description=description,
                long_description=None,
                examples=[],
                args=args_doc,
                returns=None,
                raises=[],
                blank_after_short_description=True,
            )
    else:
        if not inspect.isroutine(func):
            if not inspect.isroutine(callable_object.__call__):
                raise TypeError(
                    f'{callable_object!r}.__call__ is not a method.'
                )
            func = callable_object.__call__
        callable_type = (
            CallableType.METHOD
            if inspect.ismethod(func)
            else CallableType.FUNCTION
        )
        if auto_doc:
            docstr = _docstr.docstr(func)
        sig = inspect.signature(func)

    module_name = getattr(func, '__module__', None)
    return cls.from_signature(
        sig=sig,
        name=func.__name__,
        qualname=func.__qualname__,
        callable_type=callable_type,
        module_name=module_name or 'wrapper',
        auto_typing=auto_typing,
        docstr=docstr,
        parent_module=sys.modules[module_name] if module_name else None,
    )

from_signature classmethod

from_signature(
    sig: Signature,
    name: str,
    callable_type: CallableType,
    module_name: str | None = None,
    qualname: str | None = None,
    auto_typing: bool = False,
    docstr: str | DocStr | None = None,
    parent_module: ModuleType | None = None,
) -> Signature

Returns PyGX signature from Python signature.

Parameters:

Name Type Description Default
sig Signature

Python signature.

required
name str

Name of the entity (class name or function/method name).

required
callable_type CallableType

the type of this callable.

required
module_name str | None

Module name of the entity.

None
qualname str | None

(Optional) qualified name of the entity.

None
auto_typing bool

If True, automatically convert argument annotations to PyGX ValueSpec objects. Otherwise use pg.typing.Any() with annotations.

False
docstr str | DocStr | None

(Optional) DocStr for this entity.

None
parent_module ModuleType | None

(Optional) Parent module from where the signature is derived. This is useful to infer classes with forward declarations.

None

Returns:

Type Description
Signature

A PyGX Signature object.

Source code in pygx/typing/_callable_signature.py
@classmethod
def from_signature(
    cls,
    sig: inspect.Signature,
    name: str,
    callable_type: CallableType,
    module_name: str | None = None,
    qualname: str | None = None,
    auto_typing: bool = False,
    docstr: str | _docstr.DocStr | None = None,
    parent_module: types.ModuleType | None = None,
) -> 'Signature':
    """Returns PyGX signature from Python signature.

    Args:
      sig: Python signature.
      name: Name of the entity (class name or function/method name).
      callable_type: the type of this callable.
      module_name: Module name of the entity.
      qualname: (Optional) qualified name of the entity.
      auto_typing: If True, automatically convert argument annotations
        to PyGX ValueSpec objects. Otherwise use pg.typing.Any()
        with annotations.
      docstr: (Optional) DocStr for this entity.
      parent_module: (Optional) Parent module from where the signature is
        derived. This is useful to infer classes with forward declarations.

    Returns:
      A PyGX Signature object.
    """
    args = []
    kwonly_args = []
    varargs = None
    varkw = None

    if isinstance(docstr, str):
        docstr = _docstr.DocStr.parse(docstr)

    def make_arg_spec(param: inspect.Parameter) -> Argument:
        """Makes argument spec from inspect.Parameter."""
        docstr_arg = docstr.parameter(param) if docstr else None
        return Argument.from_parameter(
            param,
            description=docstr_arg.description if docstr_arg else None,
            auto_typing=auto_typing,
            parent_module=parent_module,
        )

    positional_only = inspect.Parameter.POSITIONAL_ONLY
    positional_or_keyword = inspect.Parameter.POSITIONAL_OR_KEYWORD
    keyword_only = inspect.Parameter.KEYWORD_ONLY
    var_positional = inspect.Parameter.VAR_POSITIONAL
    for param in sig.parameters.values():
        arg_spec = make_arg_spec(param)
        kind = param.kind
        if kind is positional_only or kind is positional_or_keyword:
            args.append(arg_spec)
        elif kind is keyword_only:
            kwonly_args.append(arg_spec)
        elif kind is var_positional:
            varargs = arg_spec
        else:
            assert kind is inspect.Parameter.VAR_KEYWORD, kind
            varkw = arg_spec

    return_value = None
    if sig.return_annotation is not inspect.Parameter.empty:
        return_value = class_schema.ValueSpec.from_annotation(
            sig.return_annotation,
            auto_typing=auto_typing,
            parent_module=parent_module,
        )

    return cls(
        callable_type=callable_type,
        name=name,
        module_name=module_name,
        qualname=qualname,
        description=docstr.short_description if docstr else None,
        args=args,
        kwonlyargs=kwonly_args,
        varargs=varargs,
        varkw=varkw,
        return_value=return_value,
    )

make_function

make_function(
    body: list[str],
    exec_globals: dict[str, Any] | None = None,
    exec_locals: dict[str, Any] | None = None,
)

Makes a function with current signature.

Source code in pygx/typing/_callable_signature.py
def make_function(
    self,
    body: list[str],
    exec_globals: dict[str, Any] | None = None,
    exec_locals: dict[str, Any] | None = None,
):
    """Makes a function with current signature."""
    if exec_globals is None:
        exec_globals = {}
    if exec_locals is None:
        exec_locals = {}

    # Hoist locals so the per-arg loops below don't pay for
    # attribute/global lookups per iteration. `make_function` is
    # called per `pg.Object` subclass; cumulative time across class
    # creation is meaningful.
    missing = topology.MISSING_VALUE
    args = []
    args_append = args.append

    def _append_arg(
        arg_name: str,
        arg_spec: class_schema.ValueSpec,
        force_missing_as_default: bool = False,
        arg_prefix: str = '',
    ):
        # Avoid building an intermediate list per arg; concat the
        # signature fragment with a couple of string `+`s.
        sig = arg_prefix + arg_name
        annotation = arg_spec.annotation
        if annotation is not missing:
            ann_key = '_annotation_' + arg_name
            sig += ': ' + ann_key
            exec_locals[ann_key] = annotation
        if not arg_prefix and (
            force_missing_as_default or arg_spec.has_default
        ):
            def_key = '_default_' + arg_name
            sig += ' = ' + def_key
            exec_locals[def_key] = arg_spec.default
        args_append(sig)

    has_previous_default = False
    # Build positional arguments.
    for arg in self.args:
        spec = arg.value_spec
        _append_arg(arg.name, spec, has_previous_default)
        if spec.has_default:
            has_previous_default = True

    # Build variable positional arguments.
    if self.varargs:
        varargs_spec = self.varargs.value_spec
        assert isinstance(varargs_spec, vs.List), self.varargs
        _append_arg(
            self.varargs.name,
            varargs_spec.element.value,
            arg_prefix='*',
        )
    elif self.kwonlyargs:
        args_append('*')

    # Build keyword-only arguments.
    for arg in self.kwonlyargs:
        _append_arg(arg.name, arg.value_spec)

    # Build variable keyword arguments.
    if self.varkw:
        varkw_spec = self.varkw.value_spec
        assert isinstance(varkw_spec, vs.Dict), self.varkw
        varkw_schema = varkw_spec.schema
        assert varkw_schema is not None
        assert varkw_schema.dynamic_field is not None
        _append_arg(
            self.varkw.name,
            varkw_schema.dynamic_field.value,
            arg_prefix='**',
        )

    # Generate function.
    fn = coding.make_function(
        self.name,
        args=args,
        body=body,
        exec_globals=exec_globals,
        exec_locals=exec_locals,
        return_type=getattr(
            self.return_value, 'annotation', coding.NO_TYPE_ANNOTATION
        ),
    )
    fn.__module__ = self.module_name
    fn.__name__ = self.name
    fn.__qualname__ = self.qualname
    return fn

Field

Field(
    key_spec: KeySpec | str,
    value_spec: ValueSpec,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
    origin: type[Any] | None = None,
    *,
    enable_repr: bool | None = None,
    enable_compare: bool | None = None,
    enable_hash: bool | None = None,
    enable_init: bool = True,
    enable_clone: bool | None = None,
    enable_validation: bool | None = None,
    assign: str | None = None,
    alias: str | None = None
)

Bases: Formattable, WireConvertible

Class that represents the definition of one or a group of attributes.

Field is held by a pygx.Schema object for defining the name(s), the validation and transformation rules on its/their value(s) for a single symbolic attribute or a set of symbolic attributes.

A Field is defined by a tuple of 4 items:

(key specification, value specification, doc string, field metadata)

For example:

(pg.typing.StrKey('foo.*'),
 pg.typing.Int(),
 'Attributes with foo',
 {'user_data': 'bar'})

The key specification (or KeySpec, class pygx.KeySpec) and value specification (or ValueSpec, class pygx.ValueSpec) are required, while the doc string and the field metadata are optional. The KeySpec defines acceptable identifiers for this field, and the ValueSpec defines the attribute's value type, its default value, validation rules and etc. The doc string serves as the description for the field, and the field metadata can be used for attribute-based code generation.

Field supports extension, which allows the subclass to add more restrictions to a field inherited from the base class, or override its default value. A field can be frozen so that subclasses can no longer extend it.

See pygx.KeySpec and pygx.ValueSpec for details.

Constructor.

Parameters:

Name Type Description Default
key_spec KeySpec | str

Key specification of the field. Can be a string or a KeySpec instance.

required
value_spec ValueSpec

Value specification of the field.

required
description str | None

Description of the field.

None
metadata dict[str, Any] | None

A dict of objects as metadata for the field.

None
origin type[Any] | None

The class that this field originates from.

None
enable_repr bool | None

Whether this field appears in pg.Object.format / repr. None (default) resolves at read time to the field's "natural" choice: True for symbolic fields (enable_init=True), False for non-symbolic fields (enable_init=False, treated as private state). Set an explicit True / False to override.

None
enable_compare bool | None

Whether this field participates in symbolic equality. Same None resolution as enable_repr.

None
enable_hash bool | None

Whether this field participates in symbolic hashing. Same None resolution as enable_repr.

None
enable_init bool

If False, the field is excluded from the generated __init__ and routed outside the symbolic state — stored on self.__dict__ rather than in _sym_attributes — making it suitable for derived/computed state populated by on_sym_ready. Mirrors dataclasses.field(init=...). This flag is the storage-mode pivot and so does not itself accept None.

True
enable_clone bool | None

Only meaningful when enable_init=False (no-op for symbolic fields, which always clone). Same None resolution as enable_repr: None → don't copy, the cloned instance starts from the default (or unset) and on_sym_ready recomputes. Set True to preserve the source value on the clone (deep- or shallow-copied per the clone's deep flag).

None
enable_validation bool | None

Whether values are validated against the field's value spec. Same None resolution: True for symbolic (ValueSpec.apply runs at __init__ time), False for non-symbolic (values stored verbatim). Set True on a non-symbolic field to revalidate the default and every self.x = ... assignment; set False on a symbolic field to skip apply at __init__.

None
assign str | None

How this field takes the value it is given. 'adopt' — take it into this tree when it is free, and copy it when it already has a holder. Requires topo=True: a flat class adopts nothing, so the policy would be a silent no-op there and is rejected at class creation. 'clone' — ALWAYS store a copy, held or not, so this field never aliases the caller's object. 'ref' — share one node with the other holders rather than copying. None (default) — undeclared, so a second hold raises SharedHoldError and the caller resolves it explicitly (pg.maybe_clone / pg.maybe_ref). Declaring the policy here rather than at each call site removes the repetition and keeps the copy fused into adoption, where it is free. 'clone' / 'ref' are meaningful in both modes.

None
alias str | None

Optional wire-layer alias for this field's key (#517): accepted by from_json, emitted by to_json(by_alias=True) and JSON Schema. Python attribute access and __init__ kwargs always use the field name.

None

Raises:

Type Description
ValueError

metadata is not a dict.

Source code in pygx/typing/_class_schema.py
def __init__(
    self,
    key_spec: KeySpec | str,
    value_spec: ValueSpec,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
    origin: type[Any] | None = None,
    *,
    enable_repr: bool | None = None,
    enable_compare: bool | None = None,
    enable_hash: bool | None = None,
    enable_init: bool = True,
    enable_clone: bool | None = None,
    enable_validation: bool | None = None,
    assign: str | None = None,
    alias: str | None = None,
) -> None:
    """Constructor.

    Args:
      key_spec: Key specification of the field. Can be a string or a KeySpec
        instance.
      value_spec: Value specification of the field.
      description: Description of the field.
      metadata: A dict of objects as metadata for the field.
      origin: The class that this field originates from.
      enable_repr: Whether this field appears in ``pg.Object.format`` /
        ``repr``. ``None`` (default) resolves at read time to the
        field's "natural" choice: ``True`` for symbolic fields
        (``enable_init=True``), ``False`` for non-symbolic fields
        (``enable_init=False``, treated as private state). Set an
        explicit ``True`` / ``False`` to override.
      enable_compare: Whether this field participates in symbolic
        equality. Same ``None`` resolution as ``enable_repr``.
      enable_hash: Whether this field participates in symbolic
        hashing. Same ``None`` resolution as ``enable_repr``.
      enable_init: If False, the field is excluded from the generated
        ``__init__`` and routed *outside* the symbolic state — stored
        on ``self.__dict__`` rather than in ``_sym_attributes`` —
        making it suitable for derived/computed state populated by
        ``on_sym_ready``. Mirrors ``dataclasses.field(init=...)``. This
        flag is the storage-mode pivot and so does not itself accept
        ``None``.
      enable_clone: Only meaningful when ``enable_init=False`` (no-op
        for symbolic fields, which always clone). Same ``None``
        resolution as ``enable_repr``: ``None`` → don't copy, the
        cloned instance starts from the default (or unset) and
        ``on_sym_ready`` recomputes. Set ``True`` to preserve the source
        value on the clone (deep- or shallow-copied per the clone's
        ``deep`` flag).
      enable_validation: Whether values are validated against the
        field's value spec. Same ``None`` resolution: ``True`` for
        symbolic (``ValueSpec.apply`` runs at ``__init__`` time),
        ``False`` for non-symbolic (values stored verbatim). Set
        ``True`` on a non-symbolic field to revalidate the default
        and every ``self.x = ...`` assignment; set ``False`` on a
        symbolic field to skip ``apply`` at ``__init__``.
      assign: How this field takes the value it is given.
        ``'adopt'`` — take it into this tree when it is free, and copy
        it when it already has a holder. Requires ``topo=True``: a
        flat class adopts nothing, so the policy would be a silent
        no-op there and is rejected at class creation.
        ``'clone'`` — ALWAYS store a copy, held or not, so this field
        never aliases the caller's object.
        ``'ref'`` — share one node with the other holders rather than
        copying.
        ``None`` (default) — undeclared, so a second hold raises
        ``SharedHoldError`` and the caller resolves it explicitly
        (``pg.maybe_clone`` / ``pg.maybe_ref``).
        Declaring the policy here rather than at each call site removes
        the repetition and keeps the copy fused into adoption, where it
        is free. ``'clone'`` / ``'ref'`` are meaningful in both modes.
      alias: Optional wire-layer alias for this field's key (#517):
        accepted by ``from_json``, emitted by ``to_json(by_alias=True)``
        and JSON Schema. Python attribute access and ``__init__``
        kwargs always use the field name.

    Raises:
      ValueError: metadata is not a dict.
    """
    if isinstance(key_spec, str):
        key_spec = KeySpec.from_str(key_spec)
    assert isinstance(key_spec, KeySpec), key_spec
    self._key = key_spec
    self._value = value_spec
    self._description = description
    self._origin = origin
    self._enable_repr = enable_repr
    self._enable_compare = enable_compare
    self._enable_hash = enable_hash
    self._enable_init = enable_init
    self._enable_clone = enable_clone
    self._enable_validation = enable_validation
    self._assign = assign
    self._validator: Callable[[Any], None] | None = None
    self._alias: str | None = None
    if alias is not None:
        self.set_alias(alias)

    if metadata and not isinstance(metadata, dict):
        raise ValueError('metadata must be a dict.')
    self._metadata = metadata or {}

description property

description: str | None

Description of this field.

key property

key: KeySpec

Key specification of this field.

value property

value: ValueSpec

Value specification of this field.

annotation property

annotation: Any

Type annotation for this field.

metadata property

metadata: dict[str, Any]

Metadata of this field.

Metadata is defined as a dict type, so multiple annotations can be attached to a field and retrieved by key:

field = pg.typing.Field('x', pg.typing.Int(), 'An int field.',
                        {'userdata': 1})
assert field.metadata.get('userdata', None) == 1

Returns:

Type Description
dict[str, Any]

Metadata of this field as a dict.

origin property

origin: type[Any] | None

The class that this field originates from.

enable_repr property

enable_repr: bool

Whether to include this field in pg.Object.format / repr.

None resolves to enable_init (symbolic → True, non-symbolic → False).

enable_compare property

enable_compare: bool

Whether to include this field in symbolic equality comparisons.

None resolves to enable_init (symbolic → True, non-symbolic → False).

enable_hash property

enable_hash: bool

Whether to include this field in symbolic hashing.

None resolves to enable_init (symbolic → True, non-symbolic → False).

enable_init property

enable_init: bool

Whether to include this field in the generated __init__.

enable_clone property

enable_clone: bool

Whether a non-symbolic field is copied during sym_clone.

Only consulted when enable_init is False (a no-op otherwise: symbolic fields always participate in sym_clone). None resolves to enable_init — i.e. False for non-symbolic, which means the clone re-seeds from the default / leaves the field unset and on_sym_ready recomputes.

enable_validation property

enable_validation: bool

Whether values flow through ValueSpec.apply for type checking.

None resolves to enable_init (symbolic → True, non-symbolic → False).

assign property

assign: str | None

How this field takes its value: 'adopt' / 'clone' / 'ref' / None.

'adopt' takes a free value as-is and copies a held one (topo=True only); 'clone' always copies; 'ref' shares one node. None means UNDECLARED — a second hold raises SharedHoldError so the caller picks. It is deliberately not a silent default: that is the ambiguity the error exists to remove.

validator property

validator: Callable[[Any], None] | None

User validator called on the final value apply produces.

alias property

alias: str | None

Wire-layer alias for this field's key (#517).

Honored by from_json (the alias key is accepted for this field), to_json(by_alias=True) (emitted in place of the field name), and JSON Schema emission. Python attribute access and __init__ kwargs always use the field name.

default_value property

default_value: Any

Returns the default value.

frozen property

frozen: bool

Returns True if current field's value is frozen.

from_annotation classmethod

from_annotation(
    key: str | KeySpec,
    annotation: Any,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
    auto_typing=True,
    parent_module: ModuleType | None = None,
) -> Field

Gets a Field from annotation.

Source code in pygx/typing/_class_schema.py
@classmethod
def from_annotation(  # pragma: no cover
    cls,
    key: str | KeySpec,
    annotation: Any,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
    auto_typing=True,
    parent_module: types.ModuleType | None = None,
) -> 'Field':
    """Gets a Field from annotation."""
    del key, annotation, description, metadata, auto_typing, parent_module
    assert False, 'Overridden in `annotation_conversion.py`.'

set_description

set_description(description: str) -> None

Sets the description for this field.

Source code in pygx/typing/_class_schema.py
def set_description(self, description: str) -> None:
    """Sets the description for this field."""
    self._description = description

_own_value_spec

_own_value_spec() -> None

Copy-on-write ownership of the value spec (#535 review B1/M3).

pg.field(value_spec=...) embeds the user's spec object verbatim, and a spec may be shared with another class or already latched into another class's fast-path metadata — never mutate a spec this field does not own. Deep-clones the spec and swaps it in; fields themselves are per-class copies, so the swap is safe. Any writer that materializes CLASS-dependent state onto the spec (strictness, no-wrap default formalization) must take ownership first.

Source code in pygx/typing/_class_schema.py
def _own_value_spec(self) -> None:
    """Copy-on-write ownership of the value spec (#535 review B1/M3).

    `pg.field(value_spec=...)` embeds the user's spec object verbatim, and
    a spec may be shared with another class or already latched into
    another class's fast-path metadata — never mutate a spec this field
    does not own. Deep-clones the spec and swaps it in; fields themselves
    are per-class copies, so the swap is safe. Any writer that
    materializes CLASS-dependent state onto the spec (strictness, no-wrap
    default formalization) must take ownership first.
    """
    self._value = copy.deepcopy(self._value)

set_origin

set_origin(origin: type[Any]) -> None

Sets the origin (source class) of this field.

Source code in pygx/typing/_class_schema.py
def set_origin(self, origin: type[Any]) -> None:
    """Sets the origin (source class) of this field."""
    self._origin = origin

set_validator

set_validator(validator: Callable[[Any], None] | None) -> Self

Sets the user validator callable and returns self.

The validator is the field-level after check: Field.apply calls it with the final value — after the value spec accepted (and possibly coerced/wrapped) it — and it rejects by raising. Its return value is ignored: transformation is transform's job (the before hook on the value spec), so validation and coercion stay separate concerns. Exposed as a setter so pg.field(validator=...) can attach it to annotation-style fields.

Source code in pygx/typing/_class_schema.py
def set_validator(self, validator: Callable[[Any], None] | None) -> Self:
    """Sets the user validator callable and returns ``self``.

    The validator is the field-level *after* check: ``Field.apply``
    calls it with the final value — after the value spec accepted
    (and possibly coerced/wrapped) it — and it rejects by raising.
    Its return value is ignored: transformation is ``transform``'s
    job (the *before* hook on the value spec), so validation and
    coercion stay separate concerns. Exposed as a setter so
    ``pg.field(validator=...)`` can attach it to annotation-style
    fields.
    """
    if validator is not None and not callable(validator):
        raise TypeError(
            f'`validator` must be a single-arg callable. '
            f'Encountered: {validator!r}.'
        )
    self._validator = validator
    return self

set_alias

set_alias(alias: str | None) -> Self

Sets the wire-layer alias and returns self.

Exposed as a setter so pg.field(alias=...) can attach it to annotation-style fields. Cross-field checks (uniqueness, no collision with sibling field names or the _type wire key) run where all fields are known — at class creation.

Source code in pygx/typing/_class_schema.py
def set_alias(self, alias: str | None) -> Self:
    """Sets the wire-layer alias and returns ``self``.

    Exposed as a setter so ``pg.field(alias=...)`` can attach it to
    annotation-style fields. Cross-field checks (uniqueness, no
    collision with sibling field names or the ``_type`` wire key)
    run where all fields are known — at class creation.
    """
    if alias is not None and (not isinstance(alias, str) or not alias):
        raise TypeError(
            f'`alias` must be a non-empty string. Encountered: {alias!r}.'
        )
    self._alias = alias
    return self

extend

extend(base_field: Field) -> Self

Extend current field based on a base field.

Source code in pygx/typing/_class_schema.py
def extend(self, base_field: 'Field') -> Self:
    """Extend current field based on a base field."""
    self.key.extend(base_field.key)
    # `ValueSpec.extend` usually mutates `self` and returns it, but
    # may also return a *different* spec (e.g. a frozen `Literal[X]`
    # extending `Enum([..,X,..])` returns the parent `Enum` frozen
    # at `X`; `pg.typing.Inherit` returns a copy of the parent spec
    # with the child's default). Assign back so the child field
    # picks up whichever spec ``extend`` chose.
    self._value = self._value.extend(base_field.value)
    if not self._description:
        self._description = base_field.description
    if base_field.metadata:
        metadata = copy.copy(base_field.metadata)
        metadata.update(self.metadata)
        self._metadata = metadata
    # `enable_*` flags inherit from the base when this field omits them
    # (``None``). An *explicit* child value wins — including re-enabling a
    # flag the base disabled — while a child that stays silent (``None``)
    # inherits the base's ``False`` so a parent's "hide me" intent isn't
    # accidentally re-enabled by an unrelated override. The ``self._x is
    # None`` guard is what distinguishes "child re-specified" from "child
    # silent"; without it an explicit child ``True`` would be clobbered by
    # an inherited ``False`` (e.g. a ``topo=True`` subclass of a ``topo=False``
    # base could never re-wrap an inherited field). ``enable_init`` has no
    # ``None`` state (it is the storage-mode pivot), so a base's ``False``
    # always wins there — re-declaring a field doesn't silently pull a
    # base's computed/non-init field back into ``__init__``.
    # pylint: disable=protected-access
    if self._enable_repr is None and base_field._enable_repr is False:
        self._enable_repr = False
    if self._enable_compare is None and base_field._enable_compare is False:
        self._enable_compare = False
    if self._enable_hash is None and base_field._enable_hash is False:
        self._enable_hash = False
    if not base_field._enable_init:
        self._enable_init = False
    if self._enable_clone is None and base_field._enable_clone is False:
        self._enable_clone = False
    if (
        self._enable_validation is None
        and base_field._enable_validation is False
    ):
        self._enable_validation = False
    # A silent child inherits the base's validator (the `transform`
    # inheritance rule on `ValueSpec.extend`, one level up) — and its
    # alias likewise.
    if self._validator is None:
        self._validator = base_field._validator
    if self._alias is None:
        self._alias = base_field._alias
    # pylint: enable=protected-access
    return self

apply

apply(
    value: Any,
    allow_partial: bool = False,
    transform_fn: None | Callable[[KeyPath, Field, Any], Any] = None,
    root_path: KeyPath | None = None,
    inplace: bool = True,
) -> Any

Applies this field to a value, validating and completing it.

Parameters:

Name Type Description Default
value Any

Value to validate against this spec.

required
allow_partial bool

Whether partial value is allowed. This is for dict or nested dict values.

False
transform_fn None | Callable[[KeyPath, Field, Any], Any]

Function to transform applied value into final value.

None
root_path KeyPath | None

Key path for root.

None
inplace bool

If False, containers are validated on a copy so the caller's input is left untouched. Defaults to True.

True

Returns:

Type Description
Any

final value.

Any

When allow_partial is set to False (default), only fully qualified value

Any

is acceptable. When allow_partial is set to True, missing fields will

Any

be placeheld using MISSING_VALUE.

Raises:

Type Description
KeyError

if additional key is found in value, or required key is missing and allow_partial is set to False.

TypeError

if type of value is not the same as spec required.

ValueError

if value is not acceptable, or value is MISSING_VALUE while allow_partial is set to False, or the field's validator rejected the final value.

Source code in pygx/typing/_class_schema.py
def apply(
    self,
    value: Any,
    allow_partial: bool = False,
    transform_fn: None
    | (Callable[[topology.KeyPath, 'Field', Any], Any]) = None,
    root_path: topology.KeyPath | None = None,
    inplace: bool = True,
) -> Any:
    """Applies this field to a value, validating and completing it.

    Args:
      value: Value to validate against this spec.
      allow_partial: Whether partial value is allowed. This is for dict or
        nested dict values.
      transform_fn: Function to transform applied value into final value.
      root_path: Key path for root.
      inplace: If False, containers are validated on a copy so the
        caller's input is left untouched. Defaults to True.

    Returns:
      final value.
      When allow_partial is set to False (default), only fully qualified value
      is acceptable. When allow_partial is set to True, missing fields will
      be placeheld using MISSING_VALUE.

    Raises:
      KeyError: if additional key is found in value, or required key is missing
        and allow_partial is set to False.
      TypeError: if type of value is not the same as spec required.
      ValueError: if value is not acceptable, or value is MISSING_VALUE while
        allow_partial is set to False, or the field's ``validator``
        rejected the final value.
    """
    value = self._value.apply(
        value,
        allow_partial=allow_partial,
        child_transform=transform_fn,
        root_path=root_path,
        inplace=inplace,
    )

    if transform_fn:
        value = transform_fn(root_path or topology.KeyPath(), self, value)
    # The field-level *after* check: the validator sees the FINAL value —
    # post spec-accept, post wrap/adopt (`transform_fn` product) — including
    # an accepted `None`. A `MissingValue` placeholder (partial construct)
    # skips it: there is no value to judge yet; the completing write
    # re-enters `apply` and validates then.
    if self._validator is not None and not isinstance(
        value, topology.MissingValue
    ):
        try:
            self._validator(value)
        except Exception as e:  # pylint: disable=broad-except
            raise e.__class__(
                topology.message_on_path(
                    str(e), root_path or topology.KeyPath()
                )
            ).with_traceback(sys.exc_info()[2])
    return value

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this field into a string.

Source code in pygx/typing/_class_schema.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this field into a string."""
    return formatting.kvlist_str(
        [
            ('key', self._key, None),
            ('value', self._value, None),
            ('description', self._description, None),
            ('metadata', self._metadata, {}),
            ('origin', self._origin, None),
            ('enable_repr', self._enable_repr, None),
            ('enable_compare', self._enable_compare, None),
            ('enable_hash', self._enable_hash, None),
            ('enable_init', self._enable_init, True),
            ('enable_clone', self._enable_clone, None),
            ('enable_validation', self._enable_validation, None),
            ('validator', self._validator, None),
            ('alias', self._alias, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

ForwardRef

ForwardRef(module: ModuleType, qualname: str)

Bases: Formattable

Forward type reference.

Source code in pygx/typing/_class_schema.py
def __init__(self, module: types.ModuleType, qualname: str):
    self._module = module
    self._qualname = qualname
    self._resolved_cls: type[Any] | None = None
    # Only track refs that don't already resolve — module-level
    # targets resolve via getattr and never need late binding.
    if self._module_lookup() is None:
        ForwardRef._pending.setdefault((module, qualname), []).append(
            weakref.ref(self)
        )

module property

module: ModuleType

Returns the module where the name is being referenced.

name property

name: str

Returns the name of the type reference.

qualname property

qualname: str

Returns the qualified name of the reference.

type_id property

type_id: str

Returns the type id of the reference.

resolved property

resolved: bool

Returns True if the symbol for the name is resolved.

cls property

cls: type[Any]

Returns the resolved reference class.

bind

bind(cls: type[Any]) -> None

Resolve this ForwardRef instance to cls.

Per-instance: equal-but-distinct refs are not affected. Used by Schema to wire local classes into their own self-references and to satisfy sibling cross-references once both endpoints exist.

Source code in pygx/typing/_class_schema.py
def bind(self, cls: type[Any]) -> None:
    """Resolve this `ForwardRef` instance to ``cls``.

    Per-instance: equal-but-distinct refs are not affected. Used by
    `Schema` to wire local classes into their own self-references
    and to satisfy sibling cross-references once both endpoints exist.
    """
    self._resolved_cls = cls

as_annotation

as_annotation() -> type[Any] | str

Returns the forward reference as an annotation.

Source code in pygx/typing/_class_schema.py
def as_annotation(self) -> type[Any] | str:
    """Returns the forward reference as an annotation."""
    return self.cls if self.resolved else self.qualname

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_class_schema.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('module', self.module.__name__, None),
            ('name', self.qualname, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

KeySpec

Bases: Formattable, WireConvertible

Interface for key specifications.

A key specification determines what keys are acceptable for a symbolic field (see pygx.Field). Usually, symbolic attributes have a 1:1 relationship with symbolic fields. But in some cases (e.g. a dict with dynamic keys), a field can be used to describe a group of symbolic attributes:

# A dictionary that accepts key 'x' with float value
# or keys started with 'foo' with int values.
d = pg.Dict({'x': 1.0, 'foo1': 1}, value_spec=pg.typing.Dict([
   ('x', pg.typing.Float(min_value=0.0)),
   (pg.typing.StrKey('foo.*'), pg.typing.Int())
]))

You may have noticed that the code above passes a string 'x' as the key spec for a field definition. The string is automatically converted to pygx.typing.ConstStrKey.

PyGX's builtin key specifications are:

KeySpec type Class
Fixed string identifier pygx.typing.ConstStrKey
Dynamic string identifier pygx.typing.StrKey
Key of a list pygx.typing.ListKey
Key of a tuple pygx.typing.TupleKey

In most scenarios, the user uses either a string or a StrKey as the key spec, while other KeySpec subclasses (e.g. ListKey and TupleKey) are used internally to constrain list size and tuple items.

is_const abstractmethod property

is_const: bool

Returns whether current key is const.

const_text property

const_text: str | None

For const string keys, returns the literal key text; else None.

Default returns None. ConstStrKey overrides to return its text. Schema's per-field precompute paths read this instead of casting to a concrete key type — avoids any dependency on _key_specs.

match abstractmethod

match(key: Any) -> bool

Returns whether current key specification can match a key.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def match(self, key: Any) -> bool:
    """Returns whether current key specification can match a key."""

extend abstractmethod

extend(base: KeySpec) -> Self

Extend base key specification and returns self.

NOTE(daiyip): When a Field extends a base Field (from a base schema), it calls extend on both its KeySpec and ValueSpec. KeySpec.extend is to determine whether the Field key is allowed to be extended, and ValueSpec.extend is to determine the final ValueSpec after extension.

Parameters:

Name Type Description Default
base KeySpec

A base KeySpec object.

required

Returns:

Type Description
Self

A KeySpec object derived from this key spec by extending the base.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def extend(self, base: 'KeySpec') -> Self:
    """Extend base key specification and returns self.

    NOTE(daiyip): When a ``Field`` extends a base Field (from a base schema),
    it calls ``extend`` on both its ``KeySpec`` and ``ValueSpec``.
    ``KeySpec.extend`` is to determine whether the ``Field`` key is allowed to
    be extended, and ``ValueSpec.extend`` is to determine the final
    ``ValueSpec`` after extension.

    Args:
      base: A base ``KeySpec`` object.

    Returns:
      A ``KeySpec`` object derived from this key spec by extending the base.
    """

from_str classmethod

from_str(key: str) -> Self

Creates a const string key spec from a string key.

Source code in pygx/typing/_class_schema.py
@classmethod
def from_str(cls, key: str) -> Self:  # pragma: no cover
    """Creates a const string key spec from a string key."""
    del key
    assert False, 'Overridden in `key_specs.py`.'

Schema

Schema(
    fields: list[Field],
    name: str | None = None,
    base_schema_list: list[Schema] | None = None,
    description: str | None = None,
    *,
    allow_nonconst_keys: bool = False,
    metadata: dict[str, Any] | None = None,
    for_cls: type[Any] | None = None
)

Bases: Formattable, WireConvertible

Class that represents a schema.

PyGX's runtime type system is based on the concept of Schema ( class pygx.Schema), which defines what symbolic attributes are held by a symbolic type (e.g. a symbolic dict, a symbolic list or a symbolic class) and what values each attribute accepts. A Schema object consists of a list of Field (class pygx.Field), which define the acceptable keys and their values for these attributes. A Schema object is usually created automatically and associated with a symbolic type upon its declaration, through dataclass-like field annotations on pygx.Object or decorators such as pygx.symbolize or pygx.functor. For example:

class A(pg.Object):
  x: int = 1
  y: float | None = None

print(A.__schema__)

@pg.symbolize([
    ('a', pg.typing.Int()),
    ('b', pg.typing.Float())
])
def foo(a, b):
  return a + b

print(foo.__schema__)

Implementation-wise it holds an ordered dictionary of a field key (pygx.KeySpec) to its field definition (pygx.Field). The key specification describes what keys/attributes are acceptable for the field, and value specification within the Field describes the value type of the field and their validation rules, default values, and etc.

Symbolic attributes can be inherited during subclassing. Accordingly, the schema that defines a symbolic class' attributes can be inherited too by its subclasses. The fields from the bases' schema will be carried over into the subclasses' schema, while the subclass can override, by redefining that field with the same key. The subclass cannot override its base classes' field with arbitrary value specs, it must be overriding non-frozen fields with more restrictive validation rules of the same type, or change their default values. See pygx.ValueSpec.extend for more details.

The code snippet below illustrates schema inheritance during subclassing:

class A(pg.Object):
  x: typing.Annotated[
      int, pg.field(value_spec=pg.typing.Int(min_value=1))
  ]
  y: float

class B(A):
  # Further restrict inherited 'x' by specifying the max value, as well
  # as providing a default value.
  x: typing.Annotated[
      int, pg.field(value_spec=pg.typing.Int(max_value=5, default=2))
  ]
  z: typing.Annotated[
      str, pg.field(value_spec=pg.typing.Str().freeze('foo'))
  ]

assert list(B.__schema__.fields.keys()) == ['x', 'y', 'z']

class C(B):
  # Raises: 'z' is frozen in class B and cannot be extended further.
  z: str

With a schema, an input dict can be validated and completed by the schema via apply. A key that matches no field raises KeyError; a missing required field raises unless allow_partial is set to True, in which case a partially validated/transformed dict is returned, with missing values placeheld by pg.MISSING_VALUE.

Constructor.

Parameters:

Name Type Description Default
fields list[Field]

A list of Field as the definition of the schema. The order of the fields will be preserved.

required
name str | None

Optional name of this schema. Useful for debugging.

None
base_schema_list list[Schema] | None

List of schema used as base. When present, fields from these schema will be copied to this schema. Fields from the latter schema will override those from the former ones.

None
description str | None

Optional str as the description for the schema.

None
allow_nonconst_keys bool

Whether immediate fields can use non-const keys.

False
metadata dict[str, Any] | None

Optional dict of user objects as schema-level metadata.

None
for_cls type[Any] | None

Optional class that this schema applies to.

None

Raises:

Type Description
TypeError

Argument fields is not a list.

KeyError

If a field name contains characters ('.') which is not allowed, or a field name from fields already exists in parent schema.

ValueError

When failed to create ValueSpec from fields. It could be an unsupported value type, default value doesn't conform with value specification, etc.

Source code in pygx/typing/_class_schema.py
def __init__(
    self,
    fields: list[Field],
    name: str | None = None,
    base_schema_list: list['Schema'] | None = None,
    description: str | None = None,
    *,
    allow_nonconst_keys: bool = False,
    metadata: dict[str, Any] | None = None,
    for_cls: type[Any] | None = None,
):
    """Constructor.

    Args:
      fields: A list of Field as the definition of the schema. The order of the
        fields will be preserved.
      name: Optional name of this schema. Useful for debugging.
      base_schema_list: List of schema used as base. When present, fields
        from these schema will be copied to this schema. Fields from the
        latter schema will override those from the former ones.
      description: Optional str as the description for the schema.
      allow_nonconst_keys: Whether immediate fields can use non-const keys.
      metadata: Optional dict of user objects as schema-level metadata.
      for_cls: Optional class that this schema applies to.

    Raises:
      TypeError: Argument `fields` is not a list.
      KeyError: If a field name contains characters ('.') which is not
        allowed, or a field name from `fields` already exists in parent
        schema.
      ValueError: When failed to create ValueSpec from `fields`.
        It could be an unsupported value type, default value doesn't conform
        with value specification, etc.
    """
    if not isinstance(fields, list):
        raise TypeError(
            f"Argument 'fields' must be a list. Encountered: {fields}."
        )

    self._name = name
    self._allow_nonconst_keys = allow_nonconst_keys
    # `KeySpec.__hash__` / `__eq__` make str keys lookup-compatible with
    # `KeySpec` keys; widen the static type so `__getitem__(str | KeySpec)`
    # below type-checks without per-call ignores.
    self._fields: dict[Any, 'Field'] = {f.key: f for f in fields}
    self._description = description
    self._metadata = metadata or {}

    if for_cls is not None:
        for f in fields:
            if f.origin is None:
                f.set_origin(for_cls)
        _register_forward_ref_target(fields, for_cls)

    self._dynamic_field = None
    for f in fields:
        if not f.key.is_const:
            self._dynamic_field = f
            break

    self._rebuild_apply_caches()

    if base_schema_list:
        base = Schema.merge(base_schema_list)
        self.extend(base)

    if not allow_nonconst_keys and self._dynamic_field is not None:
        raise ValueError(
            f'NonConstKey is not allowed in schema. '
            f"Encountered '{self._dynamic_field.key}'."
        )

    # `pg.typing.Inherit` markers must be absorbed during ``self.extend``
    # by a parent field with the same key. Any that remain here have no
    # parent to inherit from — fail at class creation rather than letting
    # the marker silently leak into ``apply`` / serialization.
    for field in self._fields.values():
        if getattr(field._value, '_is_inherit_marker', False):
            raise TypeError(
                f'Field {field.key!r}: `pg.typing.Inherit` requires a '
                f'parent field with the same key, but none was found '
                f'in the base schema(s).'
            )

description property

description: str | None

Returns the description for the schema.

dynamic_field property

dynamic_field: Field | None

Returns the field that matches multiple keys if any.

name property

name: str | None

Name of this schema.

allow_nonconst_keys property

allow_nonconst_keys: bool

Returns whether to allow non-const keys.

fields property

fields: dict[KeySpec, Field]

Returns fields of this schema.

metadata property

metadata: dict[str, Any]

Returns metadata of this schema.

merge classmethod

merge(
    schema_list: Sequence[Schema],
    name: str | None = None,
    description: str | None = None,
) -> Schema

Merge multiple schemas into one.

For fields shared by multiple schemas, the first appearance wins, unless a later field's origin class is a subclass of the earlier one's — then the more derived definition takes over.

Parameters:

Name Type Description Default
schema_list Sequence[Schema]

A list of schemas to merge.

required
name str | None

(Optional) name of the merged schema.

None
description str | None

(Optional) description of the schema.

None

Returns:

Type Description
Schema

The merged schema.

Source code in pygx/typing/_class_schema.py
@classmethod
def merge(
    cls,
    schema_list: Sequence['Schema'],
    name: str | None = None,
    description: str | None = None,
) -> 'Schema':
    """Merge multiple schemas into one.

    For fields shared by multiple schemas, the first appearance wins,
    unless a later field's ``origin`` class is a subclass of the
    earlier one's — then the more derived definition takes over.

    Args:
      schema_list: A list of schemas to merge.
      name: (Optional) name of the merged schema.
      description: (Optional) description of the schema.

    Returns:
      The merged schema.
    """
    fields = {}
    kw_field = None
    for schema in schema_list:
        for key, field in schema.fields.items():
            if key.is_const:
                if key not in fields or (
                    field.origin is not None
                    and fields[key].origin is not None
                    and issubclass(field.origin, fields[key].origin)
                ):
                    fields[key] = field
            elif kw_field is None:
                kw_field = field

    if kw_field is not None:
        fields[kw_field.key] = kw_field
    return Schema(
        list(fields.values()),
        name=name,
        description=description,
        allow_nonconst_keys=True,
    )

extend

extend(base: Schema) -> Self

Extend current schema based on a base schema.

Source code in pygx/typing/_class_schema.py
def extend(self, base: 'Schema') -> Self:
    """Extend current schema based on a base schema."""

    def _merge_field(
        path, parent_field: Field, child_field: Field
    ) -> Field:
        """Merge function on field with the same key."""
        if parent_field != topology.MISSING_VALUE:
            if topology.MISSING_VALUE == child_field:
                if (
                    not self._allow_nonconst_keys
                    and not parent_field.key.is_const
                ):
                    hints = formatting.kvlist_str(
                        [('base', base.name, None), ('path', path, None)]
                    )
                    raise ValueError(
                        f'Non-const key {parent_field.key} is not allowed to be '
                        f'added to the schema. ({hints})'
                    )
                return copy.deepcopy(parent_field)
            else:
                try:
                    child_field.extend(parent_field)
                except Exception as e:  # pylint: disable=broad-except
                    hints = formatting.kvlist_str(
                        [('base', base.name, None), ('path', path, None)]
                    )
                    raise e.__class__(f'{e} ({hints})').with_traceback(
                        sys.exc_info()[2]
                    )
        return child_field

    self._fields = topology.merge([base.fields, self.fields], _merge_field)
    self._metadata = topology.merge([base.metadata, self.metadata])

    # Inherit dynamic field from base if it's not present in the child.
    if self._dynamic_field is None:
        for k, f in self._fields.items():
            if not k.is_const:
                self._dynamic_field = f
                break

    self._rebuild_apply_caches()
    return self

is_compatible

is_compatible(other: Schema) -> bool

Returns whether current schema is compatible with the other schema.

NOTE(daiyip): schema A is compatible with schema B when: schema A and schema B have the same keys, with compatible values specs.

Parameters:

Name Type Description Default
other Schema

Other schema.

required

Returns:

Type Description
bool

True if values acceptable to the other schema are acceptable to the current schema.

Raises: TypeError: If other is not a schema object.

Source code in pygx/typing/_class_schema.py
def is_compatible(self, other: 'Schema') -> bool:
    """Returns whether current schema is compatible with the other schema.

    NOTE(daiyip): schema A is compatible with schema B when:
    schema A and schema B have the same keys, with compatible values specs.

    Args:
      other: Other schema.

    Returns:
      True if values acceptable to the other schema are acceptable to
        the current schema.
    Raises:
      TypeError: If `other` is not a schema object.
    """
    if not isinstance(other, Schema):
        raise TypeError(
            f"Argument 'other' should be a Schema object. "
            f'Encountered {other}.'
        )

    for key_spec in other.keys():
        if key_spec not in self:
            return False

    for key_spec, field in self.items():
        if key_spec not in other:
            return False
        if not field.value.is_compatible(other[key_spec].value):
            return False
    return True

get_field

get_field(key: str | int) -> Field | None

Get field definition (Field) for a key.

Parameters:

Name Type Description Default
key str | int

Key to look up — a string key name, or an int (e.g. a list index matched against a ListKey).

required

Returns:

Type Description
Field | None

Matched field. A field is considered a match when: * Its key spec is a ConstStrKey that equals to the input key. * Or it's the first field whose key spec is a NonConstKey which matches the input key.

Source code in pygx/typing/_class_schema.py
def get_field(self, key: str | int) -> Field | None:
    """Get field definition (Field) for a key.

    Args:
      key: Key to look up — a string key name, or an int (e.g. a list
        index matched against a ``ListKey``).

    Returns:
      Matched field. A field is considered a match when:
        * Its key spec is a ConstStrKey that equals to the input key.
        * Or it's the first field whose key spec is a NonConstKey
          which matches the input key.
    """
    # For const string key, we can directly retrieve from fields dict.
    # ConstStrKey hashes/compares equal to its str, so str lookups work.
    if key in self._fields:
        return self._fields[key]

    if self._allow_nonconst_keys:
        for key_spec, field in self._fields.items():
            if key_spec.match(key):
                return field
    return None

set_description

set_description(description: str) -> None

Sets the description for the schema.

Source code in pygx/typing/_class_schema.py
def set_description(self, description: str) -> None:
    """Sets the description for the schema."""
    self._description = description

resolve

resolve(
    keys: Iterable[str | int],
) -> tuple[dict[KeySpec, list[str]], list[str | int]]

Resolve keys by grouping them by their matched fields.

Parameters:

Name Type Description Default
keys Iterable[str | int]

A list of string keys.

required

Returns:

Type Description
tuple[dict[KeySpec, list[str]], list[str | int]]

A tuple of matched key results and unmatched keys. Matched key results are an ordered dict of KeySpec to matched keys, in field declaration order. Unmatched keys are strings from input.

Source code in pygx/typing/_class_schema.py
def resolve(
    self, keys: Iterable[str | int]
) -> tuple[dict[KeySpec, list[str]], list[str | int]]:
    """Resolve keys by grouping them by their matched fields.

    Args:
      keys: A list of string keys.

    Returns:
      A tuple of matched key results and unmatched keys.
        Matched key results are an ordered dict of KeySpec to matched keys,
        in field declaration order.
        Unmatched keys are strings from input.
    """
    fields = self._fields
    # Fast path: all-const schema (the common case). The matched_keys
    # dict is precomputed at construction; resolve only needs to find
    # unmatched input keys. Get_value handles absent keys via MISSING.
    if not self._has_nonconst_keys:
        unmatched_keys = [key for key in keys if key not in fields]
        assert self._const_matched_keys_template is not None
        return (self._const_matched_keys_template, unmatched_keys)

    keys = list(keys)
    input_keyset = set(keys)
    nonconst_key_specs = self._nonconst_key_specs
    nonconst_keys = {k: [] for k in nonconst_key_specs}
    unmatched_keys = []
    keys_by_key_spec = dict()

    for key in keys:
        if key not in fields:
            matched_nonconst_keys = False
            for key_spec in nonconst_key_specs:
                if key_spec.match(key):
                    nonconst_keys[key_spec].append(key)
                    matched_nonconst_keys = True
                    break
            if not matched_nonconst_keys:
                unmatched_keys.append(key)

    for key_spec in fields.keys():
        keys = []
        if not key_spec.is_const:
            keys = nonconst_keys.get(key_spec, [])
        elif key_spec in input_keyset:
            ck_text = key_spec.const_text
            assert ck_text is not None
            keys.append(ck_text)
        keys_by_key_spec[key_spec] = keys

    return (keys_by_key_spec, unmatched_keys)

apply

apply(
    dict_obj: dict[str, Any],
    allow_partial: bool = False,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
    root_path: KeyPath | None = None,
    respect_raw_storage: bool | None = None,
    inplace: bool = True,
) -> dict[str, Any]

Apply this schema to a dict object, validate and transform it.

Parameters:

Name Type Description Default
dict_obj dict[str, Any]

JSON dict type that (maybe) conform to the schema.

required
allow_partial bool

Whether allow partial object to be created.

False
child_transform None | Callable[[KeyPath, Field, Any], Any]

Function to transform child node values in dict_obj into their final values. Transform function is called on leaf nodes first, then on their containers, recursively. The signature of transform_fn is: (path, field, value) -> new_value Argument path is a KeyPath object to the field. Argument field is on which Field the value should apply. Argument value is the value from input that matches a Field from the schema, with child fields already transformed by this function. There are possible values for these two arguments:

------------------------------------------------------------
                        |   field       | value
------------------------------------------------------------
The value with          |               |
applicable Field is     |   Not None    | Not MISSING_VALUE
found in schema.        |               |
value.                  |               |
------------------------------------------------------------
The value is            |               |
not present for a       |   Not None    | MISSING_VALUE
key defined in schema.  |               |
------------------------------------------------------------

Return value will be inserted to the parent dict under path, unless return value is MISSING_VALUE.

None
root_path KeyPath | None

KeyPath of root element of dict_obj.

None
respect_raw_storage bool | None

Per-field gating of child_transform by declared storage. None (default): no gating — the transform applies uniformly to every field. True / False: each const field keeps or skips the transform per its value spec's declared storage — a spec declaring symbolic storage (e.g. a pg.Dict[...] / pg.List[...] annotation) keeps it, a spec declaring raw storage (a builtin dict[...] / list[...] annotation) skips it, and a spec that declares neither falls back to this value (symbolic object constructs thread their class's topo setting here, #740). Wildcard (non-const) fields are exempt: extra keys always keep the transform. Only consulted on the resolve path, which is the only path a child_transform takes.

None
inplace bool

When True (default), validate/transform dict_obj in place and return it (the historical behavior). When False, leave the caller's input untouched. This level is validated on a shallow copy, and inplace=False threads down the recursion so each nested container that apply would mutate is likewise copied at its own level. Verbatim values apply does not recurse into (an Object, an Any payload, a scalar) are NOT copied — they keep their identity.

True

Returns:

Type Description
dict[str, Any]

A dict filled by the schema with transformed values.

Raises:

Type Description
KeyError

Key is not allowed in schema.

TypeError

Type of dict values are not aligned with schema.

ValueError

Value of dict values are not aligned with schema.

Source code in pygx/typing/_class_schema.py
def apply(
    self,
    dict_obj: dict[str, Any],
    allow_partial: bool = False,
    child_transform: None
    | (Callable[[topology.KeyPath, Field, Any], Any]) = None,
    root_path: topology.KeyPath | None = None,
    respect_raw_storage: bool | None = None,
    inplace: bool = True,
) -> dict[str, Any]:  # pyformat: disable
    # pyformat: disable
    """Apply this schema to a dict object, validate and transform it.

    Args:
      dict_obj: JSON dict type that (maybe) conform to the schema.
      allow_partial: Whether allow partial object to be created.
      child_transform: Function to transform child node values in dict_obj into
        their final values. Transform function is called on leaf nodes first,
        then on their containers, recursively.
        The signature of transform_fn is: `(path, field, value) -> new_value`
        Argument `path` is a KeyPath object to the field. Argument `field` is
        on which Field the value should apply. Argument `value` is the value
        from input that matches a Field from the schema, with child fields
        already transformed by this function.
        There are possible values for these two arguments:

        ```text
        ------------------------------------------------------------
                                |   field       | value
        ------------------------------------------------------------
        The value with          |               |
        applicable Field is     |   Not None    | Not MISSING_VALUE
        found in schema.        |               |
        value.                  |               |
        ------------------------------------------------------------
        The value is            |               |
        not present for a       |   Not None    | MISSING_VALUE
        key defined in schema.  |               |
        ------------------------------------------------------------
        ```

        Return value will be inserted to the parent dict under path, unless
        return value is MISSING_VALUE.
      root_path: KeyPath of root element of dict_obj.
      respect_raw_storage: Per-field gating of ``child_transform`` by
        declared storage. ``None`` (default): no gating — the transform
        applies uniformly to every field. ``True`` / ``False``: each
        const field keeps or skips the transform per its value spec's
        declared storage — a spec declaring symbolic storage (e.g. a
        ``pg.Dict[...]`` / ``pg.List[...]`` annotation) keeps it, a
        spec declaring raw storage (a builtin ``dict[...]`` /
        ``list[...]`` annotation) skips it, and a spec that declares
        neither falls back to this value (symbolic object constructs
        thread their class's ``topo`` setting here, #740). Wildcard
        (non-const) fields are exempt: extra keys always keep the
        transform. Only consulted on the resolve path, which is the
        only path a ``child_transform`` takes.
      inplace: When True (default), validate/transform ``dict_obj`` in place
        and return it (the historical behavior). When False, leave the
        caller's input untouched. This level is validated on a shallow copy,
        and `inplace=False` threads down the recursion so each nested
        container that apply would mutate is likewise copied at its own
        level. Verbatim values apply does not recurse into (an ``Object``, an
        ``Any`` payload, a scalar) are NOT copied — they keep their identity.

    Returns:
      A dict filled by the schema with transformed values.

    Raises:
      KeyError: Key is not allowed in schema.
      TypeError: Type of dict values are not aligned with schema.
      ValueError: Value of dict values are not aligned with schema.
    """  # pyformat: enable
    # Pure-return (validation-v2 §5): `inplace=False` shallow-copies THIS
    # dict so the top-level fill/validate does not touch the caller's dict;
    # `inplace` then threads through the per-field recursion (Field.apply →
    # ValueSpec.apply → nested Schema.apply / List._apply) so a nested
    # container is copied exactly where apply would mutate it, while a value
    # apply stores verbatim keeps its identity (no over-copy).
    if not inplace:
        dict_obj = copy.copy(dict_obj)
    # All Schema.apply fast paths require a plain dict and no per-field
    # transform. Falls through to the generic resolve-based path otherwise.
    # `result` is captured (not returned) so `inplace=False` can re-emit it in
    # SCHEMA order below (results are always dicts, so `None` = "not yet set").
    result = None
    if child_transform is None and type(dict_obj) is dict:
        # Path 1 (fastest): every field is fast-eligible AND the dict
        # already has exactly the right keys with correct types.
        bulk = self._const_bulk_check
        if (
            bulk is not None
            and len(dict_obj) == len(bulk)
            and all(type(dict_obj.get(k)) is t for k, t in bulk)
        ):
            result = dict_obj
        else:
            # Path 2 (mixed-spec partial-fast): bulk-verify the fast-eligible
            # field subset in two C calls, then iterate just the slow fields.
            if (
                self._const_partial_fast_getter is not None
                and self._const_slow_fields
            ):
                result = self._apply_const_partial_fast(
                    dict_obj, allow_partial, root_path, inplace
                )
            # Path 3 (all-const, plain dict): per-field iteration with inline
            # fast-eligible short-circuit. Handles any const-keyed schema
            # without going through `resolve()`.
            if result is None and self._const_apply_fields is not None:
                result = self._apply_const_iterate(
                    dict_obj, allow_partial, root_path, inplace
                )

    # Slow path: nonconst keys, symbolic dicts, or child_transform set.
    # Goes through `resolve()` for full key-matching support.
    if result is None:
        result = self._apply_via_resolve(
            dict_obj,
            allow_partial,
            child_transform,
            root_path,
            respect_raw_storage,
            inplace,
        )

    # `inplace=False` returns a FRESH dict in normalized (schema) order — const
    # fields in declaration order, then non-const extras — so it is the pure
    # producer `validate_fields` used to be.
    return result if inplace else self._reorder_to_schema(result)

validate

validate(
    dict_obj: dict[str, Any],
    allow_partial: bool = False,
    root_path: KeyPath | None = None,
) -> None

Validates whether the dict object conforms to the schema.

Source code in pygx/typing/_class_schema.py
def validate(
    self,
    dict_obj: dict[str, Any],
    allow_partial: bool = False,
    root_path: topology.KeyPath | None = None,
) -> None:
    """Validates whether the dict object conforms to the schema."""
    self.apply(
        copy.deepcopy(dict_obj),
        allow_partial=allow_partial,
        root_path=root_path,
    )

set_name

set_name(name: str) -> None

Sets the name of this schema.

Source code in pygx/typing/_class_schema.py
def set_name(self, name: str) -> None:
    """Sets the name of this schema."""
    self._name = name

get

get(key: str | KeySpec, default: Field | None = None) -> Field | None

Returns field by key with default value if not found.

Source code in pygx/typing/_class_schema.py
def get(
    self, key: str | KeySpec, default: Field | None = None
) -> Field | None:
    """Returns field by key with default value if not found."""
    return self._fields.get(key, default)

keys

keys() -> Iterable[KeySpec]

Returns an iterable of KeySpecs in declaration order.

Source code in pygx/typing/_class_schema.py
def keys(self) -> Iterable[KeySpec]:
    """Returns an iterable of KeySpecs in declaration order."""
    return self._fields.keys()

values

values() -> Iterable[Field]

Returns an iterable of Field in declaration order.

Source code in pygx/typing/_class_schema.py
def values(self) -> Iterable[Field]:
    """Returns an iterable of Field in declaration order."""
    return self._fields.values()

items

items() -> Iterable[tuple[KeySpec, Field]]

Returns an iterable of (KeySpec, Field) tuple in declaration order.

Source code in pygx/typing/_class_schema.py
def items(self) -> Iterable[tuple[KeySpec, Field]]:
    """Returns an iterable of (KeySpec, Field) tuple in declaration order."""
    return self._fields.items()

format

format(
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    *,
    cls_name: str | None = None,
    bracket_type: BracketType = ROUND,
    fields_only: bool = False,
    **kwargs
) -> str

Format current Schema into nicely printed string.

Source code in pygx/typing/_class_schema.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    *,
    cls_name: str | None = None,
    bracket_type: formatting.BracketType = formatting.BracketType.ROUND,
    # pylint: disable-next=unused-argument
    fields_only: bool = False,
    **kwargs,
) -> str:
    """Format current Schema into nicely printed string."""
    return formatting.kvlist_str(
        [
            ('name', self.name, None),
            ('description', self.description, None),
            ('fields', list(self.fields.values()), []),
            ('allow_nonconst_keys', self.allow_nonconst_keys, True),
            ('metadata', self.metadata, {}),
        ],
        label=cls_name or self.__class__.__name__,
        bracket_type=bracket_type,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

to_json_schema

to_json_schema(
    *,
    include_type_name: bool = True,
    include_subclasses: bool = False,
    inline_nested_refs: bool = False,
    **kwargs
) -> dict[str, Any]

Converts this field to JSON schema.

Source code in pygx/typing/_class_schema.py
def to_json_schema(  # pragma: no cover
    self,
    *,
    include_type_name: bool = True,
    include_subclasses: bool = False,
    inline_nested_refs: bool = False,
    **kwargs,
) -> dict[str, Any]:
    """Converts this field to JSON schema."""
    del include_type_name, include_subclasses, inline_nested_refs, kwargs
    assert False, 'Overridden in `json_schema.py`.'

from_json_schema classmethod

from_json_schema(
    json_schema: dict[str, Any],
    class_fn: Callable[[str, Schema], type[Any]] | None = None,
    add_json_schema_as_metadata: bool = False,
) -> Schema

Converts a JSON schema to a schema.

Source code in pygx/typing/_class_schema.py
@classmethod
def from_json_schema(  # pragma: no cover
    cls,
    json_schema: dict[str, Any],
    class_fn: Callable[[str, 'Schema'], type[Any]] | None = None,
    add_json_schema_as_metadata: bool = False,
) -> 'Schema':
    """Converts a JSON schema to a schema."""
    del json_schema, class_fn, add_json_schema_as_metadata
    assert False, 'Overridden in `json_schema.py`.'

ValueSpec

Bases: Formattable, WireConvertible

Interface for value specifications.

A value specification defines what values are acceptable for a symbolic field (see pygx.Field). When assignments take place on the attributes for the field, the associated ValueSpec object will kick in to intercept the process and take care of the following aspects:

  • Type check
  • Noneable check
  • Value validation and transformation
  • Default value assignment

See .apply for more details.

Different aspects of assignment interception are handled by the following methods:

Aspect name Property/Method
Type check .value_type
Noneable check .is_noneable
Type-specific value validation & transformation .apply
User transform .transform
Default value lookup .default

There are many ValueSpec subclasses, each correspond to a commonly used Python type, e.g. Bool, Int, Float and etc. PyGX's builtin value specifications are:

ValueSpec type Class
bool pygx.typing.Bool
int pygx.typing.Int
float pygx.typing.Float
str pygx.typing.Str
enum pygx.typing.Enum
list pygx.typing.List
tuple pygx.typing.Tuple
dict pygx.typing.Dict
instance of a class pygx.typing.Object
callable pygx.typing.Callable
functor pygx.typing.Functor
type pygx.typing.Type
union pygx.typing.Union
any pygx.typing.Any

Construction

A value specification is an instance of a ValueSpec subclass. All pygx.ValueSpec subclasses follow a common pattern to construct:

pg.typing.<ValueSpecClass>(
    [validation-rules],
    [default=<default>],
    [transform=<transform>])

After creation, a ValueSpec object can be modified with chaining. The code below creates an int specification with default value 1 and can accept None:

pg.typing.Int().noneable().set_default(1)

Usage

To apply a value specification on an user input to get the accepted value, pygx.ValueSpec.apply shall be used:

value = pg.typing.Int(min_value=1).apply(4)
assert value == 4

Extension

Besides, a ValueSpec object can extend another ValueSpec object to obtain a more restricted ValueSpec object. For example:

pg.typing.Int(min_value=1).extend(pg.typing.Int(max_value=5))

will end up with:

pg.typing.Int(min_value=1, max_value=5)

which will be useful when subclass adds additional restrictions to an inherited symbolic attribute from its base class. For some use cases, a value spec can be frozen to avoid subclass extensions:

pg.typing.Int().freeze(1)

ValueSpec objects can be created and modified with chaining. For example:

pg.typing.Int().noneable().set_default(1)

The code above creates an int specification with default value 1 and can accept None.

On symbolic classes, ValueSpec objects are usually derived from plain Python type annotations:

class A(pg.Object):
  a: list[str]
  b: dict = {'key': 'value'}
  c: list | None
  x: int
  y: str | None
  z: int | float

A bare ValueSpec instance is not a valid field annotation (static type checkers cannot read it). When a field needs rules an annotation cannot express, attach the spec via pg.field(value_spec=...):

class B(pg.Object):
  x: typing.Annotated[
      int, pg.field(value_spec=pg.typing.Int(min_value=0))
  ]

Note that builtin container annotations (list[str], dict) declare raw storage — assigned values stay plain list / dict — while pg.List[str] / pg.Dict[str, int] annotations declare symbolic storage.

value_type abstractmethod property

value_type: type[Any] | tuple[type[Any], ...] | None

Returns acceptable (resolved) value type(s).

fast_apply_info abstractmethod property

fast_apply_info: FastApplyInfo

Hot-path metadata describing whether apply() may be skipped.

Concrete specs return a FastApplyInfo with is_eligible=True and the exact value_type when:

  • apply(v) is a no-op for values where type(v) is value_type;
  • no transform is set;
  • the spec is not frozen;
  • neither _apply nor _validate performs work at runtime.

Otherwise specs return a FastApplyInfo() with is_eligible=False (typically a shared module-level sentinel).

Schema and container specs (List/Tuple/Dict) consult this to bypass the full apply pipeline when type identity already implies validity. See FastApplyInfo for the bulk-check helper.

forward_refs abstractmethod property

forward_refs: set[ForwardRef]

Returns forward references used by the value spec.

is_noneable abstractmethod property

is_noneable: bool

Returns True if current value spec accepts None.

default abstractmethod property

default: Any

Returns the default value.

If no default is provided, MISSING_VALUE will be returned for non-dict types. For Dict type, a dict that may contains nested MISSING_VALUE will be returned.

When a default_factory is configured (and no static default), this property returns MISSING_VALUE. Use materialize_default to obtain a per-instance value.

default_factory abstractmethod property

default_factory: Callable[[], Any] | None

Returns the zero-arg factory used to materialize defaults, or None.

has_default property

has_default: bool

Returns True if the default value is provided.

frozen abstractmethod property

frozen: bool

Returns True if current value spec is frozen.

annotation abstractmethod property

annotation: Any

Returns PyType annotation. MISSING_VALUE if annotation is absent.

transform abstractmethod property

transform: Callable[[Any], Any] | None

Returns a transform that will be applied on the input before apply.

type_resolved property

type_resolved: bool

Returns True if all forward references are resolved.

noneable abstractmethod

noneable(is_noneable=True, use_none_as_default: bool = True) -> Self

Marks none-able and returns self.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def noneable(
    self, is_noneable=True, use_none_as_default: bool = True
) -> Self:
    """Marks none-able and returns `self`."""

set_default abstractmethod

set_default(
    default: Any,
    use_default_apply: bool = True,
    root_path: KeyPath | None = None,
) -> Self

Sets the default value and returns self.

Parameters:

Name Type Description Default
default Any

Default value.

required
use_default_apply bool

If True, invoke apply to the value, otherwise use default value as is.

True
root_path KeyPath | None

(Optional) The path of the field.

None

Returns:

Type Description
Self

ValueSpec itself.

Raises:

Type Description
ValueError

If default value cannot be applied when use_default_apply is set to True.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def set_default(
    self,
    default: Any,
    use_default_apply: bool = True,
    root_path: topology.KeyPath | None = None,
) -> Self:
    """Sets the default value and returns `self`.

    Args:
      default: Default value.
      use_default_apply: If True, invoke `apply` to the value, otherwise use
        default value as is.
      root_path: (Optional) The path of the field.

    Returns:
      ValueSpec itself.

    Raises:
      ValueError: If default value cannot be applied when use_default_apply
        is set to True.
    """

set_default_factory abstractmethod

set_default_factory(default_factory: Callable[[], Any] | None) -> Self

Sets a zero-arg factory called whenever a default is materialized.

Stdlib-dataclasses-style: the factory is invoked once per default materialization (not once per spec construction), so factories like lambda: time.time() produce a fresh value each time. The factory result is validated through apply on each call.

Setting default_factory clears any static default. Setting it to None clears the factory.

Parameters:

Name Type Description Default
default_factory Callable[[], Any] | None

Zero-arg callable, or None to clear.

required

Returns:

Type Description
Self

ValueSpec itself.

Raises:

Type Description
TypeError

If default_factory is not None and not callable.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def set_default_factory(
    self,
    default_factory: Callable[[], Any] | None,
) -> Self:
    """Sets a zero-arg factory called whenever a default is materialized.

    Stdlib-``dataclasses``-style: the factory is invoked once per default
    materialization (not once per spec construction), so factories like
    ``lambda: time.time()`` produce a fresh value each time. The factory
    result is validated through `apply` on each call.

    Setting ``default_factory`` clears any static default. Setting it to
    ``None`` clears the factory.

    Args:
      default_factory: Zero-arg callable, or None to clear.

    Returns:
      ValueSpec itself.

    Raises:
      TypeError: If ``default_factory`` is not None and not callable.
    """

set_transform abstractmethod

set_transform(transform: Callable[[Any], Any] | None) -> Self

Sets the user transform callable and returns self.

The transform runs on the raw input of apply, before type validation; its result then re-enters the validation pipeline.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def set_transform(
    self,
    transform: Callable[[Any], Any] | None,
) -> Self:
    """Sets the user transform callable and returns self.

    The transform runs on the raw input of `apply`, before type
    validation; its result then re-enters the validation pipeline.
    """

materialize_default abstractmethod

materialize_default(allow_partial: bool = False) -> Any

Returns a per-instance default value.

If a default_factory is set, calls it and validates the result. Otherwise returns a selective copy of the static default (via copy_default, #664): recognized copyable shapes — builtin containers, symbolic values, dataclasses — are deep-copied per instance, while arbitrary-type templates are shared by reference. Returns MISSING_VALUE when no default is configured.

With allow_partial=True, factory-only specs return a typed MissingValue instead of invoking the factory — used by partial-apply paths (e.g. schema auto-default generation) so the factory only fires during real instance construction.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def materialize_default(self, allow_partial: bool = False) -> Any:
    """Returns a per-instance default value.

    If a ``default_factory`` is set, calls it and validates the result.
    Otherwise returns a selective copy of the static default (via
    `copy_default`, #664): recognized copyable shapes — builtin
    containers, symbolic values, dataclasses — are deep-copied per
    instance, while arbitrary-type templates are shared by reference.
    Returns ``MISSING_VALUE`` when no default is configured.

    With ``allow_partial=True``, factory-only specs return a typed
    ``MissingValue`` instead of invoking the factory — used by
    partial-apply paths (e.g. schema auto-default generation) so the
    factory only fires during real instance construction.
    """

freeze abstractmethod

freeze(
    permanent_value: Any = MISSING_VALUE, apply_before_use: bool = True
) -> Self

Sets the default value using a permanent value and freezes current spec.

A frozen value spec will not accept any value that is not the default value. A frozen value spec is useful when a subclass fixes the value of a symbolic attribute and wants to prevent it from being modified.

Parameters:

Name Type Description Default
permanent_value Any

A permanent value used for current spec. If MISSING_VALUE, freeze the value spec with current default value.

MISSING_VALUE
apply_before_use bool

If True, invoke apply on permanent value when permanent_value is provided, otherwise use it as is.

True

Returns:

Type Description
Self

ValueSpec itself.

Raises:

Type Description
ValueError

If the current default value is MISSING_VALUE and the permanent value is not specified.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def freeze(
    self,
    permanent_value: Any = topology.MISSING_VALUE,
    apply_before_use: bool = True,
) -> Self:
    """Sets the default value using a permanent value and freezes current spec.

    A frozen value spec will not accept any value that is not the default
    value. A frozen value spec is useful when a subclass fixes the value of a
    symbolic attribute and wants to prevent it from being modified.

    Args:
      permanent_value: A permanent value used for current spec.
        If MISSING_VALUE, freeze the value spec with current default value.
      apply_before_use: If True, invoke `apply` on permanent value
        when permanent_value is provided, otherwise use it as is.

    Returns:
      ValueSpec itself.

    Raises:
      ValueError: If the current default value is `MISSING_VALUE` and
        the permanent value is not specified.
    """

is_compatible abstractmethod

is_compatible(other: ValueSpec) -> bool

Returns True if values acceptable to other is acceptable to this spec.

Parameters:

Name Type Description Default
other ValueSpec

Other value spec.

required

Returns:

Type Description
bool

True if values that is applicable to the other value spec can be applied to current spec. Otherwise False.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def is_compatible(self, other: 'ValueSpec') -> bool:
    """Returns True if values acceptable to `other` is acceptable to this spec.

    Args:
      other: Other value spec.

    Returns:
      True if values that is applicable to the other value spec can be applied
        to current spec. Otherwise False.
    """

extend abstractmethod

extend(base: ValueSpec) -> Self

Extends a base spec with current spec's rules.

Parameters:

Name Type Description Default
base ValueSpec

Base ValueSpec to extend.

required

Returns:

Type Description
Self

ValueSpec itself.

Raises:

Type Description
TypeError

When this value spec cannot extend from base.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def extend(self, base: 'ValueSpec') -> Self:
    """Extends a base spec with current spec's rules.

    Args:
      base: Base ValueSpec to extend.

    Returns:
      ValueSpec itself.

    Raises:
      TypeError: When this value spec cannot extend from base.
    """

apply abstractmethod

apply(
    value: Any,
    allow_partial: bool = False,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
    root_path: KeyPath | None = None,
    inplace: bool = True,
) -> Any

Validates, completes and transforms the input value.

Here is the procedure of apply:

(1) If the spec is frozen, only the default value (or a missing
    value) is accepted; the frozen default is returned.
(2) If the input is MISSING_VALUE, raise "Required value is not
    specified" — unless `allow_partial` is True, in which case a
    MissingValue placeholder is returned.
(3) If the input is None, accept it iff the spec is noneable;
    otherwise raise ValueError.
(4) If the input is a CustomTyping instance, let its custom_apply
    intercept: it either returns the final value directly or asks
    to continue with the standard pipeline.
(5) If a user transform is set, call it on the input value; its
    result re-enters the pipeline (without the transform).
(6) Check the value type against value_spec.value_type. On a
    mismatch, attempt the implicit coercions — the universal
    int -> float widening and the target type's __pg_accept__
    classmethod — unless the spec is strict; raise TypeError when
    none accepts the value.
(7) Run type-specific validation and completion (numeric bounds,
    regex, enum membership, callable signature checks, ...).
    Container specs such as Dict, List and Tuple recurse into
    child values via their child specs.

Parameters:

Name Type Description Default
value Any

Input value to apply.

required
allow_partial bool

If True, partial value is allowed. This is useful for container types (dict, list, tuple).

False
child_transform None | Callable[[KeyPath, Field, Any], Any]

Function to transform child node values into final values. (NOTE: This transform will not be performed on current value. Instead transform on current value is done by Field.apply, which has adequate information to call transform with both KeySpec and ValueSpec).

None
root_path KeyPath | None

Key path of current node.

None
inplace bool

If False, containers are validated on a copy so the caller's input is left untouched; verbatim values apply does not recurse into keep their identity. Defaults to True.

True

Returns:

Type Description
Any

Final value:

  • When allow_partial is set to False (default), only input value that has no missing values can be applied.
  • When allow_partial is set to True, missing fields will be placeheld using MISSING_VALUE.

Raises:

Type Description
KeyError

If additional key is found in value, or required key is missing and allow_partial is set to False.

TypeError

If type of value is not the same as spec required.

ValueError

If value is not acceptable, or value is MISSING_VALUE while allow_partial is set to False.

Source code in pygx/typing/_class_schema.py
@abc.abstractmethod
def apply(
    self,
    value: Any,
    allow_partial: bool = False,
    child_transform: None
    | (Callable[[topology.KeyPath, 'Field', Any], Any]) = None,
    root_path: topology.KeyPath | None = None,
    inplace: bool = True,
) -> Any:
    """Validates, completes and transforms the input value.

    Here is the procedure of ``apply``:

    ```text
    (1) If the spec is frozen, only the default value (or a missing
        value) is accepted; the frozen default is returned.
    (2) If the input is MISSING_VALUE, raise "Required value is not
        specified" — unless `allow_partial` is True, in which case a
        MissingValue placeholder is returned.
    (3) If the input is None, accept it iff the spec is noneable;
        otherwise raise ValueError.
    (4) If the input is a CustomTyping instance, let its custom_apply
        intercept: it either returns the final value directly or asks
        to continue with the standard pipeline.
    (5) If a user transform is set, call it on the input value; its
        result re-enters the pipeline (without the transform).
    (6) Check the value type against value_spec.value_type. On a
        mismatch, attempt the implicit coercions — the universal
        int -> float widening and the target type's __pg_accept__
        classmethod — unless the spec is strict; raise TypeError when
        none accepts the value.
    (7) Run type-specific validation and completion (numeric bounds,
        regex, enum membership, callable signature checks, ...).
        Container specs such as Dict, List and Tuple recurse into
        child values via their child specs.
    ```

    Args:
      value: Input value to apply.
      allow_partial: If True, partial value is allowed. This is useful for
        container types (dict, list, tuple).
      child_transform: Function to transform child node values into final
        values.
        (NOTE: This transform will not be performed on current value. Instead
        transform on current value is done by Field.apply, which has adequate
        information to call transform with both KeySpec and ValueSpec).
      root_path: Key path of current node.
      inplace: If False, containers are validated on a copy so the
        caller's input is left untouched; verbatim values apply does not
        recurse into keep their identity. Defaults to True.

    Returns:
      Final value:

        * When allow_partial is set to False (default), only input value that
          has no missing values can be applied.
        * When allow_partial is set to True, missing fields will be placeheld
          using MISSING_VALUE.

    Raises:
      KeyError: If additional key is found in value, or required key is missing
        and allow_partial is set to False.
      TypeError: If type of value is not the same as spec required.
      ValueError: If value is not acceptable, or value is MISSING_VALUE while
        allow_partial is set to False.
    """

from_annotation classmethod

from_annotation(
    annotation: Any,
    auto_typing: bool = False,
    accept_value_as_annotation: bool = False,
    parent_module: ModuleType | None = None,
) -> ValueSpec

Gets a concrete ValueSpec from annotation.

Source code in pygx/typing/_class_schema.py
@classmethod
def from_annotation(  # pragma: no cover
    cls,
    annotation: Any,
    auto_typing: bool = False,
    accept_value_as_annotation: bool = False,
    parent_module: types.ModuleType | None = None,
) -> 'ValueSpec':
    """Gets a concrete ValueSpec from annotation."""
    del annotation, auto_typing, accept_value_as_annotation, parent_module
    assert False, 'Overridden in `annotation_conversion.py`.'

to_json_schema

to_json_schema(
    *,
    include_type_name: bool = True,
    include_subclasses: bool = False,
    inline_nested_refs: bool = False,
    **kwargs
) -> dict[str, Any]

Converts this field to JSON schema.

Source code in pygx/typing/_class_schema.py
def to_json_schema(  # pragma: no cover
    self,
    *,
    include_type_name: bool = True,
    include_subclasses: bool = False,
    inline_nested_refs: bool = False,
    **kwargs,
) -> dict[str, Any]:
    """Converts this field to JSON schema."""
    del include_type_name, include_subclasses, inline_nested_refs, kwargs
    assert False, 'Overridden in `json_schema.py`.'

from_json_schema classmethod

from_json_schema(
    json_schema: dict[str, Any],
    class_fn: Callable[[str, Schema], type[Any]] | None = None,
    add_json_schema_as_metadata: bool = False,
) -> ValueSpec

Converts a JSON schema to a value spec.

Source code in pygx/typing/_class_schema.py
@classmethod
def from_json_schema(  # pragma: no cover
    cls,
    json_schema: dict[str, Any],
    class_fn: Callable[[str, 'Schema'], type[Any]] | None = None,
    add_json_schema_as_metadata: bool = False,
) -> 'ValueSpec':
    """Converts a JSON schema to a value spec."""
    del json_schema, class_fn, add_json_schema_as_metadata
    assert False, 'Overridden in `json_schema.py`.'

CustomTyping

Interface for values that take over a field's validation.

Instances of CustomTyping subclasses can be assigned to fields of any ValueSpec, and take over ValueSpec.apply via the custom_apply method.

As a result, CustomTyping makes the schema system extensible without modifying existing value specs. For example, marker objects such as pygx.oneof extend CustomTyping so they can sit in any field regardless of its value spec.

custom_apply abstractmethod

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, Any]

Custom apply on a value based on its original value spec.

Parameters:

Name Type Description Default
path KeyPath

KeyPath of current object under its object tree.

required
value_spec ValueSpec

Original value spec for this field.

required
allow_partial bool

Whether allow partial object to be created.

required
child_transform None | Callable[[KeyPath, Field, Any], Any]

Function to transform child node values into their final values. Transform function is called on leaf nodes first, then on their parents, recursively.

None

Returns:

Type Description
tuple[bool, Any]

A tuple (proceed_with_standard_apply, value_to_proceed). If proceed_with_standard_apply is set to False, value_to_proceed will be used as final value.

Raises:

Type Description
TypeError

If the value is not compatible with the value spec.

Source code in pygx/typing/_custom_typing.py
@abc.abstractmethod
def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: class_schema.ValueSpec,
    allow_partial: bool,
    child_transform: None
    | (Callable[[topology.KeyPath, class_schema.Field, Any], Any]) = None,
) -> tuple[bool, Any]:
    """Custom apply on a value based on its original value spec.

    Args:
      path: KeyPath of current object under its object tree.
      value_spec: Original value spec for this field.
      allow_partial: Whether allow partial object to be created.
      child_transform: Function to transform child node values into their final
        values. Transform function is called on leaf nodes first, then on their
        parents, recursively.

    Returns:
      A tuple (proceed_with_standard_apply, value_to_proceed).
        If proceed_with_standard_apply is set to False, value_to_proceed
        will be used as final value.

    Raises:
      TypeError: If the value is not compatible with the value spec.
    """

DocStr dataclass

DocStr(
    style: DocStrStyle,
    short_description: str | None,
    long_description: str | None,
    examples: list[DocStrExample],
    args: dict[str, DocStrArgument],
    returns: DocStrReturns | None,
    raises: list[DocStrRaises],
    blank_after_short_description: bool = True,
)

A docstring parsed into its structured parts.

Produced by DocStr.parse / docstr — carries the description, arguments, returns, raises and examples sections as typed entries.

description property

description: str | None

Returns short_description + long_description.

parameter

parameter(param: Parameter) -> DocStrArgument | None

Returns doc str for an inspected parameter.

Source code in pygx/typing/_docstr.py
def parameter(self, param: inspect.Parameter) -> DocStrArgument | None:
    """Returns doc str for an inspected parameter."""
    if param.kind == inspect.Parameter.VAR_POSITIONAL:
        name = f'*{param.name}'
    elif param.kind == inspect.Parameter.VAR_KEYWORD:
        name = f'**{param.name}'
    else:
        name = param.name
    return self.args.get(name)

parse classmethod

parse(text: str, style: DocStrStyle | None = None) -> DocStr

Parses a docstring.

Source code in pygx/typing/_docstr.py
@classmethod
def parse(cls, text: str, style: DocStrStyle | None = None) -> 'DocStr':
    """Parses a docstring."""
    result = docstring_parser.parse(text, _to_parser_style(style))
    assert result.style is not None
    return cls(
        style=_from_parser_style(result.style),
        short_description=result.short_description,
        long_description=result.long_description,
        examples=[
            DocStrExample(description=e.description or '')
            for e in result.examples
        ],
        args={  # pylint: disable=g-complex-comprehension
            p.arg_name: DocStrArgument(
                name=p.arg_name,
                description=p.description or '',
                type_name=p.type_name,
                default=p.default,
                is_optional=p.is_optional,
            )
            for p in result.params
        },
        returns=DocStrReturns(  # pylint: disable=g-long-ternary
            name=result.returns.return_name,
            description=result.returns.description or '',
            is_yield=result.returns.is_generator,
        )
        if result.returns
        else None,
        raises=[
            DocStrRaises(
                type_name=r.type_name, description=r.description or ''
            )
            for r in result.raises
        ],
        blank_after_short_description=result.blank_after_short_description,
    )

DocStrArgument dataclass

DocStrArgument(
    description: str,
    name: str,
    type_name: str | None = None,
    default: str | None = None,
    is_optional: bool | None = None,
)

Bases: DocStrEntry

An entry in the "Args" section of a docstring.

DocStrEntry dataclass

DocStrEntry(description: str)

An entry in a docstring.

DocStrExample dataclass

DocStrExample(description: str)

Bases: DocStrEntry

An entry in the "Examples" section of a docstring.

DocStrRaises dataclass

DocStrRaises(description: str, type_name: str | None = None)

Bases: DocStrEntry

An entry in the "Raises" section of a docstring.

DocStrReturns dataclass

DocStrReturns(
    description: str,
    name: str | None = None,
    type_name: str | None = None,
    is_yield: bool = False,
)

Bases: DocStrEntry

An entry in the "Returns"/"Yields" section of a docstring.

DocStrStyle

Bases: Enum

Docstring style.

ConstStrKey

ConstStrKey(text: str)

Bases: KeySpecBase, StrKey

Class that represents a constant string key.

Example:

key = pg.typing.ConstStrKey('x')
assert key == 'x'
assert hash(key) == hash('x')

Constructor.

Parameters:

Name Type Description Default
text str

string value of this key.

required

Raises:

Type Description
KeyError

If key contains dots ('.'), which is not allowed.

Source code in pygx/typing/_key_specs.py
def __init__(self, text: str):
    """Constructor.

    Args:
      text: string value of this key.

    Raises:
      KeyError: If key contains dots ('.'), which is not allowed.
    """
    if '.' in text:
        raise KeyError(f"'.' cannot be used in key. Encountered: {text!r}.")
    super().__init__()
    self._text = text

text property

text: str

Text of this const string key.

match

match(key: Any) -> bool

Whether can match against an input key.

Source code in pygx/typing/_key_specs.py
def match(self, key: Any) -> bool:
    """Whether can match against an input key."""
    return self._text == key

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_key_specs.py
@override
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    del compact, verbose, root_indent, kwargs
    return self._text

ListKey

ListKey(min_value: int = 0, max_value: int | None = None)

Bases: NonConstKey

Class that represents key specification for a list.

Matches int keys (zero-based list indices) in [min_value, max_value) — max_value is the exclusive upper bound, i.e. the list size.

Example:

# Create a key spec for list indices in range [1, 5).
key = pg.typing.ListKey(min_value=1, max_value=5)

assert key.match(1)
assert key.match(4)
assert not key.match(5)
assert not key.match(0)

Constructor.

Parameters:

Name Type Description Default
min_value int

Minimum acceptable list index (inclusive).

0
max_value int | None

Exclusive upper bound for the list index (i.e. the list size). None or -1 means no upper bound; 0 is a real (empty) bound that matches no index.

None
Source code in pygx/typing/_key_specs.py
def __init__(self, min_value: int = 0, max_value: int | None = None):
    """Constructor.

    Args:
      min_value: Minimum acceptable list index (inclusive).
      max_value: Exclusive upper bound for the list index (i.e. the list
        size). ``None`` or ``-1`` means no upper bound; ``0`` is a real
        (empty) bound that matches no index.
    """
    super().__init__()
    self._min_value = min_value
    # `-1` (any negative) is the spelled form of "unbounded" (#773
    # ruling 6), normalized to None so every consumer has one sentinel.
    if max_value is not None and max_value < 0:
        max_value = None
    self._max_value = max_value

min_value property

min_value: int

Returns the inclusive lower bound of acceptable list indices.

max_value property

max_value: int | None

Returns the exclusive upper bound of acceptable list indices.

extend

extend(base: KeySpec) -> Self

Extend current key spec on top of base spec.

Source code in pygx/typing/_key_specs.py
def extend(self, base: KeySpec) -> typing.Self:
    """Extend current key spec on top of base spec."""
    if not isinstance(base, ListKey):
        raise TypeError(f'{self} cannot extend {base}: incompatible type.')

    if self.min_value < base.min_value:
        raise TypeError(
            f'{self} cannot extend {base}: min_value is smaller.'
        )
    if base.max_value is None:
        return self
    if self.max_value is None:
        self._max_value = base.max_value
    elif self.max_value > base.max_value:
        raise TypeError(
            f'{self} cannot extend {base}: max_value is greater.'
        )
    return self

match

match(key: Any) -> bool

Returns whether this key spec can match against input key.

Source code in pygx/typing/_key_specs.py
def match(self, key: Any) -> bool:
    """Returns whether this key spec can match against input key."""
    # `is None`, not falsy — `max_value=0` is a REAL (empty) bound that
    # matches no index, not "unbounded" (#773 ruling 6).
    return (
        isinstance(key, int)
        and (self._min_value <= key)
        and (self._max_value is None or self._max_value > key)
    )

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
)

Format this object.

Source code in pygx/typing/_key_specs.py
@override
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
):
    """Format this object."""
    del compact, verbose, root_indent, kwargs
    return (
        f'ListKey(min_value={self._min_value}, max_value={self._max_value})'
    )

NonConstKey

Bases: KeySpecBase

Base class of specification for non-const keys.

A non-const key spec matches a family of keys (e.g. all string keys matching a regular expression) rather than one fixed key.

Subclasses: pygx.typing.StrKey, pygx.typing.ListKey, pygx.typing.TupleKey.

StrKey

StrKey(regex: str | None = None)

Bases: NonConstKey

Class that represents a variable string key.

Used as the key spec of a field (e.g. in a pygx.typing.Dict schema) to declare a wildcard field that matches a family of string keys instead of a single name.

Matches any string key when no regex is given. With a regex, a key matches if re.match succeeds — the pattern is anchored at the start of the key only, so StrKey('foo') also matches 'foobar'; end the pattern with $ to require a full match. Non-string keys never match.

Example:

# Create a key spec that specifies all string keys started with 'foo'.
key = pg.typing.StrKey('foo.*')

assert key.match('foo')
assert key.match('foo1')
assert not key.match('bar')

Constructor.

Parameters:

Name Type Description Default
regex str | None

An optional regular expression. If set to None, any string value is acceptable.

None
Source code in pygx/typing/_key_specs.py
def __init__(self, regex: str | None = None):
    """Constructor.

    Args:
      regex: An optional regular expression. If set to None, any string value is
        acceptable.
    """
    super().__init__()
    self._regex = re.compile(regex) if regex else None

regex property

regex

Returns regular expression of this key spec.

match

match(key: Any) -> bool

Whether this key spec can match against input key.

Source code in pygx/typing/_key_specs.py
def match(self, key: Any) -> bool:
    """Whether this key spec can match against input key."""
    if not isinstance(key, str):
        return False
    if self._regex:
        return self._regex.match(key) is not None
    return True

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
)

Format this object.

Source code in pygx/typing/_key_specs.py
@override
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
):
    """Format this object."""
    del compact, verbose, root_indent, kwargs
    return formatting.kvlist_str(
        [('regex', getattr(self._regex, 'pattern', None), None)],
        label=self.__class__.__name__,
    )

TupleKey

TupleKey(index: int | None = None)

Bases: NonConstKey

Class that represents a key specification for tuple.

Example:

# Create a key spec that specifies item 0 of a tuple.
key = pg.typing.TupleKey(0)

assert key.match(0)
assert not key.match(1)

Constructor.

Parameters:

Name Type Description Default
index int | None

index of the tuple field that this key spec applies to. If None, this tuple value spec applies to all elements of a variable-length tuple.

None
Source code in pygx/typing/_key_specs.py
def __init__(self, index: int | None = None):
    """Constructor.

    Args:
      index: index of the tuple field that this key spec applies to.
        If None, this tuple value spec applies to all elements of a
        variable-length tuple.
    """
    super().__init__()
    self._index = index

index property

index: int | None

Returns the index of tuple field that the key applies to.

extend

extend(base: KeySpec) -> Self

Extends this key spec on top of a base spec.

Source code in pygx/typing/_key_specs.py
def extend(self, base: KeySpec) -> typing.Self:
    """Extends this key spec on top of a base spec."""
    if not isinstance(base, TupleKey):
        raise TypeError(f'{self} cannot extend {base}: incompatible type.')
    if self._index is None:
        self._index = base.index
    elif base.index is not None and base.index != self.index:
        raise KeyError(f'{self} cannot extend {base}: unmatched index.')
    return self

match

match(key: Any) -> bool

Returns whether this key spec can match against input key.

Source code in pygx/typing/_key_specs.py
def match(self, key: Any) -> bool:
    """Returns whether this key spec can match against input key."""
    return isinstance(key, int) and (
        self._index is None or self._index == key
    )

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
)

Format this object.

Source code in pygx/typing/_key_specs.py
@override
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
):
    """Format this object."""
    del compact, verbose, root_indent, kwargs
    return f'TupleKey(index={self._index})'

Date

Date(
    default: date | None = MISSING_VALUE,
    default_factory: Callable[[], date | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for datetime.date.

Lax coercion: ISO-8601 date strings. A datetime.datetime value is REJECTED (Pydantic parity — no silent time truncation), even though datetime subclasses date.

Examples:

pg.typing.Date()
pg.typing.Date(default=datetime.date(2026, 1, 1))
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: datetime.date | None = MISSING_VALUE,
    default_factory: (
        typing.Callable[[], datetime.date | None] | None
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        datetime.date,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Datetime

Datetime(
    default: datetime | None = MISSING_VALUE,
    default_factory: Callable[[], datetime | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for datetime.datetime.

Lax coercion (Pydantic parity): ISO-8601 strings, int/float POSIX timestamps (tz-aware UTC), and datetime.date (midnight). Strict mode accepts only datetime.datetime instances.

Examples:

# A required datetime.
pg.typing.Datetime()

# With a default.
pg.typing.Datetime(default=datetime.datetime(2026, 1, 1))

# ISO-8601 strings (and POSIX timestamps) coerce.
spec = pg.typing.Datetime()
assert spec.apply('2026-01-01T00:00:00') == datetime.datetime(2026, 1, 1)
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: datetime.datetime | None = MISSING_VALUE,
    default_factory: (
        typing.Callable[[], datetime.datetime | None] | None
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        datetime.datetime,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Decimal

Decimal(
    default: Decimal | None = MISSING_VALUE,
    default_factory: Callable[[], Decimal | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for decimal.Decimal.

Lax coercion: strings, ints, and floats — floats convert through their str form (1.5 → Decimal('1.5'), no binary artifacts, Pydantic parity). A non-finite coercion RESULT (NaN/inf) is rejected; an exact Decimal value is accepted verbatim.

Examples:

import decimal

pg.typing.Decimal()
pg.typing.Decimal(default=decimal.Decimal('0.00'))
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: decimal.Decimal | None = MISSING_VALUE,
    default_factory: (
        typing.Callable[[], decimal.Decimal | None] | None
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        decimal.Decimal,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Path

Path(
    default: Path | None = MISSING_VALUE,
    default_factory: Callable[[], Path | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for pathlib.Path.

Lax coercion: strings (pathlib.Path(value)).

Examples:

import pathlib

pg.typing.Path()
pg.typing.Path(default=pathlib.Path('/tmp'))
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: pathlib.Path | None = MISSING_VALUE,
    default_factory: (
        typing.Callable[[], pathlib.Path | None] | None
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        pathlib.Path,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Time

Time(
    default: time | None = MISSING_VALUE,
    default_factory: Callable[[], time | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for datetime.time.

Lax coercion: ISO-8601 time strings, and int/float seconds since midnight in [0, 86400) — producing a tz-aware UTC time, exactly as Pydantic v2 does.

Examples:

pg.typing.Time()
pg.typing.Time(default=datetime.time(12, 0))
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: datetime.time | None = MISSING_VALUE,
    default_factory: (
        typing.Callable[[], datetime.time | None] | None
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        datetime.time,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Timedelta

Timedelta(
    default: timedelta | None = MISSING_VALUE,
    default_factory: Callable[[], timedelta | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for datetime.timedelta.

Lax coercion: int/float seconds and ISO-8601 duration strings ('PT90S', 'P1DT2H', ...) — the same form the wire emits.

Examples:

pg.typing.Timedelta()
pg.typing.Timedelta(default=datetime.timedelta(seconds=30))
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: datetime.timedelta | None = MISSING_VALUE,
    default_factory: (
        typing.Callable[[], datetime.timedelta | None] | None
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        datetime.timedelta,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Uuid

Uuid(
    default: UUID | None = MISSING_VALUE,
    default_factory: Callable[[], UUID | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: _StdlibScalar

Value spec for uuid.UUID.

Lax coercion: canonical/hex UUID strings and 16-byte values.

Examples:

import uuid

pg.typing.Uuid()
pg.typing.Uuid(default=uuid.UUID(int=0))
Source code in pygx/typing/_stdlib_specs.py
def __init__(
    self,
    default: uuid.UUID | None = MISSING_VALUE,
    default_factory: typing.Callable[[], uuid.UUID | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    super().__init__(
        uuid.UUID,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Convertible

Runtime stub: Convertible[T, S] evaluates to T.

Any

Any(
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    annotation: Any = MISSING_VALUE,
    transform: None | Callable[[Any], Any] = None,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: ValueSpecBase

Value spec for any type.

Accepts values of any type, including None — an Any spec is always noneable. "Required" here means only that a value (possibly None) must be provided when the spec has no default.

Examples:

# A required value of any type.
pg.typing.Any()

# An optional value of any type, with None as its default value.
pg.typing.Any().noneable()

# A value of any type, with 1 as its default value.
pg.typing.Any(default=1)

Note

While Any type is very flexible and useful to pass though data between components, we should minimize its usage since minimal validation is performed on this type.

Constructor.

Parameters:

Name Type Description Default
default Any

(Optional) default value of this spec.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
annotation Any

(Optional) external provided type annotation.

MISSING_VALUE
transform None | Callable[[Any], Any]

(Optional) user-defined function to be called on the input of apply. It could be used as a type converter or a custom validator which may raise errors.

None
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    annotation: typing.Any = MISSING_VALUE,
    transform: None | (typing.Callable[[typing.Any], typing.Any]) = None,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      default: (Optional) default value of this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      annotation: (Optional) external provided type annotation.
      transform: (Optional) user-defined function to be called on the input
        of `apply`. It could be used as a type converter or a custom
        validator which may raise errors.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    super().__init__(
        object,
        default,
        default_factory=default_factory,
        transform=transform,
        is_noneable=True,
        frozen=frozen,
        strict=strict,
    )
    self._annotation = annotation

annotation property

annotation: Any

Returns type annotation.

is_compatible

is_compatible(other: ValueSpec) -> bool

Any is compatible with any ValueSpec.

Source code in pygx/typing/_value_specs.py
def is_compatible(self, other: ValueSpec) -> bool:
    """Any is compatible with any ValueSpec."""
    return True

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
            ('annotation', self._annotation, MISSING_VALUE),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

annotate

annotate(annotation: Any) -> Any

Set external type annotation.

Source code in pygx/typing/_value_specs.py
def annotate(self, annotation: typing.Any) -> 'Any':
    """Set external type annotation."""
    self._annotation = annotation
    return self

Bool

Bool(
    default: bool | None = MISSING_VALUE,
    default_factory: Callable[[], bool | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: PrimitiveType

Value spec for boolean type.

Accepts exactly bool values — there is no implicit coercion from int (pg.typing.Bool().apply(1) raises TypeError).

Examples:

# A required bool value.
pg.typing.Bool()

# A bool value with the default value set to True.
pg.typing.Bool(default=True)

# An optional bool value with default value set to None.
pg.typing.Bool().noneable()

# An optional bool value with default value set to True.
pg.typing.Bool(default=True).noneable()

# A frozen bool with value set to True that is not modifiable by subclasses.
pg.typing.Bool().freeze(True)

Constructor.

Parameters:

Name Type Description Default
default bool | None

Default value for the value spec.

MISSING_VALUE
default_factory Callable[[], bool | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    default: bool | None = MISSING_VALUE,
    default_factory: typing.Callable[[], bool | None] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      default: Default value for the value spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    super().__init__(
        bool,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Callable

Callable(
    args: list[ValueSpecOrAnnotation] | None = None,
    kw: None | list[tuple[str, ValueSpecOrAnnotation]] = None,
    returns: ValueSpecOrAnnotation | None = None,
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    transform: None | Callable[[Any], Callable[..., Any]] = None,
    callable_type: type[Any] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, ValueSpecBase

Value spec for callable.

A value must be callable. When args / kw / returns are provided, the value's signature is checked for compatibility at apply time: the callable must accept the declared number of positional arguments (its own *args absorbs extras) and every declared keyword name (its own **kwargs absorbs unknown names). When the callable's parameters carry value specs of their own (e.g. a functor created via pygx.symbolize with explicit field definitions), the declared arg/return specs are also checked against them; plain Python type annotations on the callable are not converted, so they are not checked. Incompatibility raises TypeError.

Examples:

# A required callable object with any args.
pg.typing.Callable()

# An optional callable objects with the first argument as int, and the
# second argument as float. The field has None as its default value.
pg.typing.Callable([pg.typing.Int(), pg.typing.Float()]).noneable()

# A callable object that has its first argument as int, and has keyword
# arguments 'x' (any type), 'y' (a str) and return value as int.
pg.typing.Callable(
    [pg.typing.Int()],
    kw=[
        ('x', pg.typing.Any()),
        ('y', pg.typing.Str())
    ],
    returns=pg.typing.Int())

See also: pygx.typing.Functor.

Constructor.

Parameters:

Name Type Description Default
args list[ValueSpecOrAnnotation] | None

(Optional) value specs for the callable's positional arguments.

None
kw None | list[tuple[str, ValueSpecOrAnnotation]]

(Optional) a list of (name, value spec) tuples for the callable's keyword arguments.

None
returns ValueSpecOrAnnotation | None

(Optional) value spec for the return value.

None
default Any

(Optional) default value of this spec.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
transform None | Callable[[Any], Callable[..., Any]]

(Optional) user-defined function to be called on the input of apply. It could be used as a type converter or a custom validator which may raise errors.

None
callable_type type[Any] | None

(Optional) a type the callable value must be an instance of (e.g. pg.Functor for Functor specs). If None, any callable is acceptable.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    args: list[ValueSpecOrAnnotation] | None = None,
    kw: None | (list[tuple[str, ValueSpecOrAnnotation]]) = None,
    returns: ValueSpecOrAnnotation | None = None,
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    transform: None
    | (
        typing.Callable[[typing.Any], typing.Callable[..., typing.Any]]
    ) = None,
    callable_type: type[typing.Any] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      args: (Optional) value specs for the callable's positional
        arguments.
      kw: (Optional) a list of (name, value spec) tuples for the
        callable's keyword arguments.
      returns: (Optional) value spec for the return value.
      default: (Optional) default value of this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      transform: (Optional) user-defined function to be called on the input
        of `apply`. It could be used as a type converter or a custom
        validator which may raise errors.
      callable_type: (Optional) a type the callable value must be an
        instance of (e.g. ``pg.Functor`` for `Functor` specs). If None,
        any callable is acceptable.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    args = args or []
    kw = kw or []
    if not isinstance(args, list):
        raise TypeError(
            f"'args' should be a list of ValueSpec objects. "
            f'Encountered: {args!r}.'
        )
    args = [
        ValueSpec.from_annotation(arg, auto_typing=True) for arg in args
    ]

    if not isinstance(kw, list):
        raise TypeError(
            f"'kw' should be a list of (name, value_spec) tuples. "
            f'Encountered: {kw!r}.'
        )
    for arg in kw:
        if (
            not isinstance(arg, tuple)
            or len(arg) != 2
            or not isinstance(arg[0], str)
        ):
            raise TypeError(
                f"'kw' should be a list of (name, value_spec) tuples. "
                f'Encountered: {kw!r}.'
            )

    kw = [
        (k, ValueSpec.from_annotation(v, auto_typing=True)) for k, v in kw
    ]

    if returns is not None:
        returns = ValueSpec.from_annotation(returns, auto_typing=True)
    self._args = args
    self._kw = kw
    self._return_value = returns
    super().__init__(
        callable_type,
        default,
        default_factory=default_factory,
        transform=transform,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

args property

args: list[ValueSpec]

Value specs for positional arguments.

kw property

kw: list[tuple[str, ValueSpec]]

Names and value specs for keyword arguments.

return_value property

return_value: ValueSpec | None

Value spec for return value.

forward_refs

forward_refs() -> set[ForwardRef]

Returns forward references used in this spec.

Source code in pygx/typing/_value_specs.py
@cached_property
@override
def forward_refs(self) -> set[class_schema.ForwardRef]:
    """Returns forward references used in this spec."""
    forward_refs = set()
    for arg in self.args:
        forward_refs.update(arg.forward_refs)
    for _, arg in self.kw:
        forward_refs.update(arg.forward_refs)
    if self.return_value:
        forward_refs.update(self.return_value.forward_refs)
    return forward_refs

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this spec.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this spec."""
    return formatting.kvlist_str(
        [
            ('args', self._args, []),
            ('kw', self._kw, []),
            ('returns', self._return_value, None),
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Dict

Dict(
    schema: (
        ValueSpec
        | Schema
        | dict[FieldKeyDef, FieldValueDef]
        | list[Field | FieldDef]
        | type[Any]
        | None
    ) = None,
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    transform: None | Callable[[Any], dict[Any, Any]] = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    *,
    symbolic: bool | None = None,
    auto_wrap: bool = True
)

Bases: Generic, ValueSpecBase

Value spec for dict type.

Accepts dict values, validated against an optional schema: with no schema, arbitrary key/value pairs are accepted; a value spec constrains the values under arbitrary string keys; a field list / dict / TypedDict class defines per-key fields — missing keys are filled from field defaults, and keys matched by no field are rejected with KeyError.

Examples:

# A required dict of arbitrary keys and values.
pg.typing.Dict()

# An optional Dict of keys started with 'foo' and int values, with None as
# the default value.
pg.typing.Dict([
    (pg.typing.StrKey(), pg.typing.Int())
]).noneable()

# A dict with two keys ('x' and 'y').
pg.typing.Dict([
    ('x', pg.typing.Float()),
    ('y', pg.typing.Int(min_value=1))
])

# A dict with a user validator.
def validate_sum(d):
  if sum(d.values()) > 1.:
    raise ValueError('The sum of the dict values should be less than 1.')

pg.typing.Dict([
    (pg.typing.StrKey(), pg.typing.Float())
], transform=validate_sum)

# A frozen dict that prevents subclass to extend/override.
pg.typing.Dict([
    ('x', 1),
    ('y', 2.0)
]).freeze()

Constructor.

Parameters:

Name Type Description Default
schema ValueSpec | Schema | dict[FieldKeyDef, FieldValueDef] | list[Field | FieldDef] | type[Any] | None

(Optional) a Schema object for this Dict, or a dict of field key to field value definition, or a list of field definitions, or a TypedDict class, or a value spec. If None, it specifies a schema-less Dict that may accept arbitrary key/value pairs. If a TypedDict class, it specifies a closed Dict whose const fields mirror the TypedDict's keys (extra keys are rejected; optional keys become noneable fields defaulting to None). If a value spec, it specifies a Dict that may accept arbitrary keys with values constrained by the value spec. A field definition is a tuple of (, , [description], [metadata]). A field value definition is a tuple of (, [description], [metadata]).

None
default Any

Default value. If MISSING_VALUE, the default value will be computed according to the schema.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
transform None | Callable[[Any], dict[Any, Any]]

(Optional) user-defined function to be called on the input of apply. It could be used as a type converter or a custom validator which may raise errors.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
symbolic bool | None

Tri-state storage declaration, normally authored by a field annotation rather than passed directly. True — this spec stores a symbolic pg.Dict (the pg.Dict[str, V] annotation form); assigned raw dicts are converted. False — the annotation explicitly named a builtin dict (dict[str, V]), so values stay raw dicts. None (the default) — no annotation authored this spec (e.g. pg.typing.Dict(...) built directly, or a schema-declared field); storage then follows the enclosing wrapping rules (see auto_wrap). Only an explicit False suppresses wrapping — None does not. The declaration applies at any nesting depth: in pg.Dict[str, dict[str, Any]] the inner spec is False, so its values stay raw dicts.

None
auto_wrap bool

Whether a raw dict/list landing in a position this spec does not constrain to a container type (e.g. an Any-valued entry) is wrapped into its symbolic form. Governs only such unannotated positions — an explicit container annotation always wins, so pg.Dict[str, dict[str, Any]] keeps its inner values raw regardless. pg.Dict[str, Any] wraps an incoming raw member dict; annotate pg.Dict[str, Any, False] to keep such members verbatim.

True
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    schema: (
        class_schema.ValueSpec
        | Schema
        | dict[class_schema.FieldKeyDef, class_schema.FieldValueDef]
        | list[Field | class_schema.FieldDef]
        | type[typing.Any]
        | None
    ) = None,  # pylint: disable=bad-whitespace
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    transform: None
    | (typing.Callable[[typing.Any], dict[typing.Any, typing.Any]]) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    *,
    symbolic: bool | None = None,
    auto_wrap: bool = True,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      schema: (Optional) a Schema object for this Dict, or a dict of field key
        to field value definition, or a list of field definitions, or a
        TypedDict class, or a value spec.
        If None, it specifies a schema-less Dict that may accept arbitrary
        key/value pairs.
        If a TypedDict class, it specifies a closed Dict whose const fields
        mirror the TypedDict's keys (extra keys are rejected; optional keys
        become noneable fields defaulting to None).
        If a value spec, it specifies a Dict that may accept arbitrary keys with
        values constrained by the value spec.
        A field definition is a tuple of
          (<key_spec>, <value_spec>, [description], [metadata]).
        A field value definition is a tuple of
          (<value_spec>, [description], [metadata]).
      default: Default value. If MISSING_VALUE, the default value will be
        computed according to the schema.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      transform: (Optional) user-defined function to be called on the input
        of `apply`. It could be used as a type converter or a custom
        validator which may raise errors.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
      symbolic: Tri-state storage declaration, normally authored by a
        field annotation rather than passed directly. `True` — this
        spec stores a symbolic `pg.Dict` (the `pg.Dict[str, V]`
        annotation form); assigned raw dicts are converted. `False` —
        the annotation explicitly named a builtin `dict`
        (`dict[str, V]`), so values stay raw dicts. `None` (the
        default) — no annotation authored this spec (e.g.
        `pg.typing.Dict(...)` built directly, or a schema-declared
        field); storage then follows the enclosing wrapping rules
        (see `auto_wrap`). Only an explicit `False` suppresses
        wrapping — `None` does not. The declaration applies at any
        nesting depth: in `pg.Dict[str, dict[str, Any]]` the inner
        spec is `False`, so its values stay raw dicts.
      auto_wrap: Whether a raw `dict`/`list` landing in a position
        this spec does not constrain to a container type (e.g. an
        `Any`-valued entry) is wrapped into its symbolic form.
        Governs only such unannotated positions — an explicit
        container annotation always wins, so
        `pg.Dict[str, dict[str, Any]]` keeps its inner values raw
        regardless. `pg.Dict[str, Any]` wraps an incoming raw member
        `dict`; annotate `pg.Dict[str, Any, False]` to keep such
        members verbatim.
    """
    if schema is not None:
        if pg_inspect.is_typeddict_annotation(schema):
            # A TypedDict (bare class or subscripted generic alias)
            # adopts the schema of its converted closed spec —
            # `Dict(TD)` ≡ `ValueSpec.from_annotation(TD)`, one source
            # of truth for the key/qualifier semantics (see
            # `_annotation_conversion._typed_dict_value_spec`).
            converted = ValueSpec.from_annotation(schema, auto_typing=True)
            schema = typing.cast('Dict', converted).schema
        elif not isinstance(schema, (Schema, list, dict)):
            schema = [
                (
                    key_specs.StrKey(),
                    ValueSpec.from_annotation(schema, auto_typing=True),
                )
            ]
        if not isinstance(schema, Schema):
            schema = class_schema.create_schema(
                schema, allow_nonconst_keys=True
            )

    self._schema = typing.cast(typing.Optional[Schema], schema)
    self._symbolic = symbolic
    self._auto_wrap = auto_wrap
    super().__init__(
        dict,
        default,
        default_factory=default_factory,
        transform=transform,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

    # Detect the open-str-key pattern: `Dict[(StrKey(), value_spec)]` (also
    # produced from `Dict(value_spec)`, `Dict[str, V]`). For this common
    # case, `Schema.apply` is overkill — there is no constant-key bookkeeping,
    # no nonconst matching against multiple key specs. The fast path
    # iterates the input dict directly, falling back to `Schema.apply` only
    # if a non-string key is encountered (to preserve unmatched-key error
    # reporting) or a per-field transform is in play.
    self._open_str_value_spec: ValueSpecBase | None = None
    if self._schema is not None and len(self._schema.fields) == 1:
        only_key, only_field = next(iter(self._schema.fields.items()))
        if (
            isinstance(only_key, key_specs.StrKey)
            and only_key.regex is None
        ):
            self._open_str_value_spec = typing.cast(
                ValueSpecBase,
                only_field._value,  # pylint: disable=protected-access
            )

    # Schema-generated default, for the ONE case the base constructor does
    # not already cover. `self._schema` is assigned ABOVE `super().__init__`,
    # so when the base calls `set_default(default)` it takes this class's
    # override, sees the schema and generates — no repeat needed. The base
    # SKIPS `set_default` only on the `default_factory` branch, and for a
    # schema'd Dict the schema-generated default supersedes the factory
    # (which is why this also clears it). Calling it unconditionally re-ran
    # `Schema.apply` on the just-generated default, firing every field's
    # `validator` a second time per class creation (#680).
    if MISSING_VALUE == default and default_factory is not None:
        self.set_default(default)

schema property

schema: Schema | None

Returns the schema of this dict spec.

symbolic property

symbolic: bool | None

Storage declaration: True stores a symbolic pg.Dict, False a raw dict, None when no annotation declared storage.

auto_wrap property

auto_wrap: bool

Returns True if raw containers in unannotated positions wrap.

noneable

noneable(is_noneable: bool = True, use_none_as_default: bool = True) -> Self

Dict-specific noneable: replaces the default with None.

Unlike the base noneable, turning noneable on sets the default to None even when a default (e.g. one generated from the schema) already exists; turning it off regenerates the schema-based default.

Source code in pygx/typing/_value_specs.py
def noneable(
    self, is_noneable: bool = True, use_none_as_default: bool = True
) -> Self:
    """Dict-specific noneable: replaces the default with ``None``.

    Unlike the base `noneable`, turning noneable on sets the default
    to ``None`` even when a default (e.g. one generated from the
    schema) already exists; turning it off regenerates the
    schema-based default.
    """
    self._is_noneable = is_noneable
    if is_noneable:
        if use_none_as_default:
            self.set_default(None, False)
    elif self._default is None:
        # Automatically generate default based on schema.
        self.set_default(MISSING_VALUE)
    return self

forward_refs

forward_refs() -> set[ForwardRef]

Returns forward references used in this spec.

Source code in pygx/typing/_value_specs.py
@cached_property
@override
def forward_refs(self) -> set[class_schema.ForwardRef]:
    """Returns forward references used in this spec."""
    if self.schema is None:
        return set()
    forward_refs = set()
    for field in self.schema.fields.values():
        forward_refs.update(field.value.forward_refs)
    return forward_refs

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            (
                'fields',
                list(self._schema.values()) if self._schema else None,
                None,
            ),
            # Visible: two specs differing ONLY in storage must not render
            # identically (§6 annotation honesty). Each omitted at its
            # DEFAULT (`symbolic` raw, `auto_wrap` on), so pre-§6 reprs
            # are unchanged.
            ('symbolic', self._symbolic, None),
            ('auto_wrap', self._auto_wrap, True),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Enum

Enum(
    values: list[Any],
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, PrimitiveType

Value spec for enum type.

An applied value must equal one of the candidate values, which may mix types. A None among the values makes the spec noneable (and noneable adds/removes None from the candidate list). The default, when given, must be one of the candidates.

Examples:

# A str enum value with options 'a', 'b', 'c' and its default set to 'a'.
pg.typing.Enum(['a', 'b', 'c'], 'a')

# A mixed-type enum value.
pg.typing.Enum(['a', 5, True], 'a')

# An optional enum value with default value set to 'a'.
pg.typing.Enum(['a', 'b', 'c'], 'a').noneable()

# A frozen enum with value set to 'a' that is not modifiable by subclasses.
pg.typing.Enum(['a', 'b', 'c'], 'a').freeze('a')

# Values outside the candidate list are rejected (raises ValueError).
# pg.typing.Enum(['a', 'b'], 'a').apply('c')

Constructor.

Parameters:

Name Type Description Default
values list[Any]

all acceptable values.

required
default Any

default value for this spec.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    values: list[typing.Any],
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    frozen: bool = False,
    strict: bool | None = None,
):
    """Constructor.

    Args:
      values: all acceptable values.
      default: default value for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    if not isinstance(values, list) or not values:
        raise ValueError(
            f'Values for Enum should be a non-empty list. Found {values!r}.'
        )
    if MISSING_VALUE != default and default not in values:
        raise ValueError(
            f'Enum default value {default!r} is not in candidate list '
            f'{values!r}.'
        )

    value_type = None
    for v in values:
        if v is None:
            continue
        if value_type is None:
            value_type = type(v)
        else:
            next_type = type(v)
            if issubclass(value_type, next_type):
                value_type = next_type
            elif not issubclass(next_type, value_type):
                value_type = None
                break

    is_noneable = any([v is None for v in values])

    # NOTE(daiyip): When enum values are strings, we relax the `value_type`
    # to accept text types (unicode) as well. This allows enum value be read
    # from unicode JSON file.
    if value_type is not None and issubclass(value_type, str):
        value_type = str
    self._values = values
    super().__init__(
        value_type,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

values property

values: list[Any]

Returns all acceptable values of this spec.

noneable

noneable(is_noneable: bool = True, use_none_as_default: bool = True) -> Self

Enum-specific noneable: edits the candidate list.

Turning noneable on appends None to values (when absent); turning it off removes None (and clears a None default).

Source code in pygx/typing/_value_specs.py
def noneable(
    self, is_noneable: bool = True, use_none_as_default: bool = True
) -> Self:
    """Enum-specific noneable: edits the candidate list.

    Turning noneable on appends ``None`` to `values` (when absent);
    turning it off removes ``None`` (and clears a ``None`` default).
    """
    if is_noneable:
        if None not in self._values:
            self._values.append(None)
    else:
        if None in self._values:
            self._values.remove(None)
            if self._default is None:
                self._default = MISSING_VALUE
    self._is_noneable = is_noneable
    return self

is_compatible

is_compatible(other: ValueSpec) -> bool

Enum specific compatibility check.

Source code in pygx/typing/_value_specs.py
def is_compatible(self, other: ValueSpec) -> bool:
    """Enum specific compatibility check."""
    if other.frozen and other.default in self.values:
        return True
    return super().is_compatible(other)

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('values', self._values, None),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Float

Float(
    default: float | None = MISSING_VALUE,
    default_factory: Callable[[], float | None] | None = None,
    min_value: float | None = None,
    max_value: float | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    gt: float | None = None,
    ge: float | None = None,
    lt: float | None = None,
    le: float | None = None,
)

Bases: Number

Value spec for float type.

Accepts float values; an int (or bool) input is widened to float unless strict=True. min_value / max_value bounds are inclusive.

Examples:

# A required float value.
pg.typing.Float()

# A required float value with min and max value (both inclusive.)
pg.typing.Float(min_value=1.0, max_value=10.0)

# A float value with the default value set to 1
pg.typing.Float(default=1)

# An optional float value with default value set to None.
pg.typing.Float().noneable()

# An optional float value with default value set to 1.0.
pg.typing.Float(default=1.0).noneable()

# A frozen float with value set to 1.0 that is not modifiable by subclasses.
pg.typing.Float().freeze(1)

Constructor.

Parameters:

Name Type Description Default
default float | None

(Optional) default value for this spec.

MISSING_VALUE
default_factory Callable[[], float | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
min_value float | None

(Optional) minimum value of acceptable values.

None
max_value float | None

(Optional) maximum value of acceptable values.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
gt float | None

Exclusive lower bound (#522, pydantic spelling) — on Int it rewrites to min_value=gt + 1; on Float it raises (not representable — use ge). Mutually exclusive with ge / min_value.

None
ge float | None

Inclusive lower bound — synonym of min_value.

None
lt float | None

Exclusive upper bound — Int rewrite max_value=lt - 1; raises on Float. Mutually exclusive with le / max_value.

None
le float | None

Inclusive upper bound — synonym of max_value.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    default: float | None = MISSING_VALUE,
    default_factory: typing.Callable[[], float | None] | None = None,
    min_value: float | None = None,
    max_value: float | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    gt: float | None = None,
    ge: float | None = None,
    lt: float | None = None,
    le: float | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      default: (Optional) default value for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      min_value: (Optional) minimum value of acceptable values.
      max_value: (Optional) maximum value of acceptable values.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
      gt: Exclusive lower bound (#522, pydantic spelling) — on Int it
        rewrites to ``min_value=gt + 1``; on Float it raises (not
        representable — use ``ge``). Mutually exclusive with
        ``ge`` / ``min_value``.
      ge: Inclusive lower bound — synonym of ``min_value``.
      lt: Exclusive upper bound — Int rewrite ``max_value=lt - 1``;
        raises on Float. Mutually exclusive with ``le`` / ``max_value``.
      le: Inclusive upper bound — synonym of ``max_value``.
    """
    min_value, max_value = _normalize_bound_aliases(
        'Float', False, min_value, max_value, gt, ge, lt, le
    )
    super().__init__(
        float,
        default,
        default_factory=default_factory,
        min_value=min_value,
        max_value=max_value,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Functor

Functor(
    args: list[ValueSpecOrAnnotation] | None = None,
    kw: None | list[tuple[str, ValueSpecOrAnnotation]] = None,
    returns: ValueSpecOrAnnotation | None = None,
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    transform: None | Callable[[Any], Callable[..., Any]] = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Callable

Value spec for Functor.

Same as pygx.typing.Callable, except that only pygx.Functor instances (created via pygx.functor or pygx.symbolize on a function) are acceptable values.

Examples:

# A required PyGX functor with any args.
pg.typing.Functor()

# An optional functor with the first argument as int, and the second
# argument as float. The field has None as its default value.
pg.typing.Functor([pg.typing.Int(), pg.typing.Float()]).noneable()

# A functor that has its first argument as int, and has keyword
# arguments 'x' (any type), 'y' (a str) and return value as int.
pg.typing.Functor(
    [pg.typing.Int()],
    kw=[
        ('x', pg.typing.Any()),
        ('y', pg.typing.Str())
    ],
    returns=pg.typing.Int())

See also: pygx.typing.Callable.

Constructor. Same as Callable.__init__, with callable_type fixed to pg.Functor.

Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    args: list[ValueSpecOrAnnotation] | None = None,
    kw: None | (list[tuple[str, ValueSpecOrAnnotation]]) = None,
    returns: ValueSpecOrAnnotation | None = None,
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    transform: None
    | (
        typing.Callable[[typing.Any], typing.Callable[..., typing.Any]]
    ) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor. Same as `Callable.__init__`, with ``callable_type``
    fixed to ``pg.Functor``.
    """
    super().__init__(
        args=args,
        kw=kw,
        returns=returns,
        default=default,
        default_factory=default_factory,
        transform=transform,
        callable_type=_functor.Functor,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Generic

Base class for value specs that support [] subscripting.

Subscripting a subclass builds a spec instance from the type arguments, giving annotation-style shorthands for common specs:

assert pg.typing.Int[0, 10] == pg.typing.Int(min_value=0, max_value=10)
assert pg.typing.Str['foo.*'] == pg.typing.Str(regex='foo.*')
assert pg.typing.Enum['a', 'b'] == pg.typing.Enum(values=['a', 'b'])

GenericMeta

Bases: ABCMeta

Metaclass that makes Generic subclasses subscriptable.

Routes Cls[...] to the class's with_type_args factory (e.g. pg.typing.Int[0, 10] becomes pg.typing.Int(min_value=0, max_value=10)).

GenericTypeAlias

GenericTypeAlias(*args, **kwargs)

Bases: Generic

Base class for generic type aliases.

A generic type alias is subscript-only: Alias[...] expands to a value spec built from other spec classes (it has no spec class of its own), and the alias itself cannot be instantiated. Subclasses: pygx.typing.Sequence, pygx.typing.Optional.

Source code in pygx/typing/_value_specs.py
def __init__(self, *args, **kwargs):
    del args, kwargs
    raise TypeError(
        f'Generic type alias {self.__class__.__name__} cannot be instantiated.'
    )

Inherit

Sentinel annotation: inherit the parent field's value_spec.

Use in a pg.Object subclass to keep the parent's value spec (including Enum / regex / numeric bounds / etc.) while overriding only the field's default value — without restating the parent's spec on the subclass. Static type checkers see Inherit as typing.Any, so the subclass's default is honored by the synthesized __init__ and the override is variance-friendly.

Example:

from typing import Annotated

class Parent(pg.Object):
  model: Annotated[
      str, pg.field(value_spec=pg.typing.Enum(['a', 'b']))
  ]

class Child(Parent):
  model: pg.typing.Inherit = 'a'  # inherits Enum, default='a'

Errors at class creation if there is no parent field with the same key, or if no default value is provided on the subclass.

Int

Int(
    default: int | None = MISSING_VALUE,
    default_factory: Callable[[], int | None] | None = None,
    min_value: int | None = None,
    max_value: int | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    gt: int | None = None,
    ge: int | None = None,
    lt: int | None = None,
    le: int | None = None,
)

Bases: Number

Value spec for int type.

Accepts int values; a bool also passes (it is an int subclass) unless strict=True. min_value / max_value bounds are inclusive; gt / ge / lt / le are pydantic-style aliases for the bounds.

Examples:

# A required int value.
pg.typing.Int()

# A required int value with min and max value (both inclusive.)
pg.typing.Int(min_value=1, max_value=10)

# A int value with the default value set to 1
pg.typing.Int(default=1)

# An optional int value with default value set to None.
pg.typing.Int().noneable()

# An optional int value with default value set to 1.
pg.typing.Int(default=1).noneable()

# A frozen int with value set to 1 that is not modifiable by subclasses.
pg.typing.Int().freeze(1)

Constructor.

Parameters:

Name Type Description Default
default int | None

(Optional) default value for this spec.

MISSING_VALUE
default_factory Callable[[], int | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
min_value int | None

(Optional) minimum value of acceptable values.

None
max_value int | None

(Optional) maximum value of acceptable values.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
gt int | None

Exclusive lower bound (#522, pydantic spelling) — on Int it rewrites to min_value=gt + 1; on Float it raises (not representable — use ge). Mutually exclusive with ge / min_value.

None
ge int | None

Inclusive lower bound — synonym of min_value.

None
lt int | None

Exclusive upper bound — Int rewrite max_value=lt - 1; raises on Float. Mutually exclusive with le / max_value.

None
le int | None

Inclusive upper bound — synonym of max_value.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    default: int | None = MISSING_VALUE,
    default_factory: typing.Callable[[], int | None] | None = None,
    min_value: int | None = None,
    max_value: int | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    gt: int | None = None,
    ge: int | None = None,
    lt: int | None = None,
    le: int | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      default: (Optional) default value for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      min_value: (Optional) minimum value of acceptable values.
      max_value: (Optional) maximum value of acceptable values.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
      gt: Exclusive lower bound (#522, pydantic spelling) — on Int it
        rewrites to ``min_value=gt + 1``; on Float it raises (not
        representable — use ``ge``). Mutually exclusive with
        ``ge`` / ``min_value``.
      ge: Inclusive lower bound — synonym of ``min_value``.
      lt: Exclusive upper bound — Int rewrite ``max_value=lt - 1``;
        raises on Float. Mutually exclusive with ``le`` / ``max_value``.
      le: Inclusive upper bound — synonym of ``max_value``.
    """
    min_value, max_value = _normalize_bound_aliases(
        'Int', True, min_value, max_value, gt, ge, lt, le
    )
    super().__init__(
        int,
        default,
        default_factory=default_factory,
        min_value=min_value,
        max_value=max_value,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

List

List(
    element_value: ValueSpecOrAnnotation,
    default: list[Any] | None = MISSING_VALUE,
    default_factory: Callable[[], list[Any] | None] | None = None,
    min_size: int | None = None,
    max_size: int | None = None,
    size: int | None = None,
    transform: None | Callable[[Any], list[Any]] = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    *,
    min_length: int | None = None,
    max_length: int | None = None,
    symbolic: bool | None = None,
    auto_wrap: bool = True
)

Bases: Generic, ValueSpecBase

Value spec for list type.

Accepts list values whose every element passes the element_value spec; optional min_size / max_size (or an exact size) bound the length.

Examples:

# A required non-negative integer list.
pg.typing.List(pg.typing.Int(min_value=0))

# An optional str list with 2-5 items.
pg.typing.List(pg.typing.Str(), min_size=2, max_size=5).noneable()

# A size-2 list of arbitrary types.
pg.typing.List(pg.typing.Any(), size=2)

# A frozen list that prevents subclass to extend/override.
pg.typing.List(pg.typing.Int()).freeze([1])

Constructor.

Parameters:

Name Type Description Default
element_value ValueSpecOrAnnotation

A ValueSpec object or an equivalent annotation as the spec for the list element.

required
default list[Any] | None

(Optional) default value for this spec.

MISSING_VALUE
default_factory Callable[[], list[Any] | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
min_size int | None

(Optional) min size of list. If None, 0 will be used.

None
max_size int | None

(Optional) max size of list.

None
size int | None

(Optional) size of List. A shortcut to specify min_size and max_size at the same time. size and min_size/max_size are mutual exclusive.

None
transform None | Callable[[Any], list[Any]]

(Optional) user-defined function to be called on the input of apply. It could be used as a type converter or a custom validator which may raise errors.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
min_length int | None

Synonym of min_size (#522, pydantic spelling). Mutually exclusive with it. Keyword-only.

None
max_length int | None

Synonym of max_size. Mutually exclusive with it. Keyword-only.

None
symbolic bool | None

Tri-state storage declaration, normally authored by a field annotation (see pygx.typing.Dict for the full semantics): True for pg.List[E] (stores a symbolic pg.List; assigned raw lists are converted), False for an explicit builtin list[E] (values stay raw lists), None (default) when no annotation authored this spec.

None
auto_wrap bool

Whether a raw dict/list landing in a position this spec does not constrain to a container type (e.g. an Any element) is wrapped into its symbolic form (see pygx.typing.Dict for the full semantics). pg.List[Any] wraps such raw elements; annotate pg.List[Any, False] to keep them verbatim.

True
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    element_value: ValueSpecOrAnnotation,
    default: list[typing.Any] | None = MISSING_VALUE,
    default_factory: typing.Callable[[], list[typing.Any] | None]
    | None = None,
    min_size: int | None = None,
    max_size: int | None = None,
    size: int | None = None,
    transform: None
    | (typing.Callable[[typing.Any], list[typing.Any]]) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    *,
    min_length: int | None = None,
    max_length: int | None = None,
    symbolic: bool | None = None,
    auto_wrap: bool = True,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      element_value: A ``ValueSpec`` object or an equivalent annotation as the
        spec for the list element.
      default: (Optional) default value for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      min_size: (Optional) min size of list. If None, 0 will be used.
      max_size: (Optional) max size of list.
      size: (Optional) size of List. A shortcut to specify min_size and max_size
        at the same time. `size` and `min_size`/`max_size` are mutual exclusive.
      transform: (Optional) user-defined function to be called on the input
        of `apply`. It could be used as a type converter or a custom
        validator which may raise errors.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
      min_length: Synonym of ``min_size`` (#522, pydantic spelling).
        Mutually exclusive with it. Keyword-only.
      max_length: Synonym of ``max_size``. Mutually exclusive with it.
        Keyword-only.
      symbolic: Tri-state storage declaration, normally authored by a
        field annotation (see [`pygx.typing.Dict`][pygx.typing.Dict]
        for the full semantics): `True` for `pg.List[E]` (stores a
        symbolic `pg.List`; assigned raw lists are converted),
        `False` for an explicit builtin `list[E]` (values stay raw
        lists), `None` (default) when no annotation authored this
        spec.
      auto_wrap: Whether a raw `dict`/`list` landing in a position
        this spec does not constrain to a container type (e.g. an
        `Any` element) is wrapped into its symbolic form (see
        [`pygx.typing.Dict`][pygx.typing.Dict] for the full
        semantics). `pg.List[Any]` wraps such raw elements; annotate
        `pg.List[Any, False]` to keep them verbatim.
    """
    min_size, max_size = _normalize_size_aliases(
        'List', min_size, max_size, min_length, max_length
    )
    element_value = ValueSpec.from_annotation(
        element_value, auto_typing=True
    )

    if size is not None and (min_size is not None or max_size is not None):
        raise ValueError(
            f'Either "size" or "min_size"/"max_size" pair can be specified. '
            f'Encountered: size={size}, min_size={min_size}, '
            f'max_size={max_size}.'
        )
    if size is not None:
        min_size = size
        max_size = size

    if min_size is None:
        min_size = 0

    if min_size < 0:
        raise ValueError(
            f'"min_size" of List must be no less than 0. '
            f'Encountered: {min_size}.'
        )
    if max_size is not None:
        if max_size < min_size:
            raise ValueError(
                f'"max_size" of List must be no less than "min_size". '
                f'Encountered: min_size={min_size}, max_size={max_size}'
            )
    self._element = Field(
        key_specs.ListKey(min_size, max_size),
        element_value,
        'Field of list element',
    )
    self._symbolic = symbolic
    self._auto_wrap = auto_wrap
    super().__init__(
        list,
        default,
        default_factory=default_factory,
        transform=transform,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )
    # Cached for use in `apply()` — saves a `self._element.value` chain
    # lookup on the hot path.
    self._element_value = element_value
    self._min_size_cached = min_size
    self._max_size_cached = max_size
    # When the element is a Dict with an all-fast-fields const schema, we
    # can verify every list entry in a single nested `all(...)` call —
    # closing the per-element Dict.apply frame.
    self._nested_dict_bulk_check = None
    if isinstance(element_value, Dict):
        ev_schema = element_value._schema
        ev_bulk = (
            ev_schema._const_bulk_check if ev_schema is not None else None
        )
        if (
            ev_bulk is not None
            and not element_value._frozen
            and element_value._transform is None
        ):
            self._nested_dict_bulk_check = (ev_bulk, len(ev_bulk))

element property

element: Field

Returns Field specification of list element.

symbolic property

symbolic: bool | None

Storage declaration: True stores a symbolic pg.List, False a raw list, None when no annotation declared storage.

auto_wrap property

auto_wrap: bool

Returns True if raw containers in unannotated positions wrap.

forward_refs property

forward_refs: set[ForwardRef]

Returns forward references used in this spec.

min_size property

min_size: int

Returns min size of the list.

max_size property

max_size: int | None

Returns max size of the list.

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('', self._element.value, None),
            # See `Dict.format` — storage is visible, defaults omitted.
            ('symbolic', self._symbolic, None),
            ('auto_wrap', self._auto_wrap, True),
            ('min_size', self.min_size, 0),
            ('max_size', self.max_size, None),
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Number

Number(
    value_type: type[int] | type[float],
    default: int | float | None = MISSING_VALUE,
    default_factory: Callable[[], int | float | None] | None = None,
    min_value: int | float | None = None,
    max_value: int | float | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, PrimitiveType

Base class for value spec of numeric types.

Constructor.

Parameters:

Name Type Description Default
value_type type[int] | type[float]

Type of number.

required
default int | float | None

Default value for this spec.

MISSING_VALUE
default_factory Callable[[], int | float | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
min_value int | float | None

(Optional) minimum acceptable value (inclusive).

None
max_value int | float | None

(Optional) maximum acceptable value (inclusive).

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    value_type: type[int] | type[float],
    default: int | float | None = MISSING_VALUE,
    default_factory: typing.Callable[[], int | float | None] | None = None,
    min_value: int | float | None = None,
    max_value: int | float | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      value_type: Type of number.
      default: Default value for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      min_value: (Optional) minimum acceptable value (inclusive).
      max_value: (Optional) maximum acceptable value (inclusive).
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    if (
        min_value is not None
        and max_value is not None
        and min_value > max_value
    ):
        raise ValueError(
            f'"max_value" must be equal or greater than "min_value". '
            f'Encountered: min_value={min_value}, max_value={max_value}.'
        )
    self._min_value = min_value
    self._max_value = max_value
    super().__init__(
        value_type,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )
    # Number._validate is a no-op when neither bound is set.
    self._validate_is_noop = min_value is None and max_value is None
    self._invalidate_fast_apply_info()  # `_validate_is_noop` finalized.

min_value property

min_value: int | float | None

Returns minimum value of acceptable values.

max_value property

max_value: int | float | None

Returns maximum value of acceptable values.

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('min', self._min_value, None),
            ('max', self._max_value, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Object

Object(
    t: type[Any] | ForwardRef | str,
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    transform: None | Callable[[Any], Any] = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, ValueSpecBase

Value spec for object type.

Accepts instances of class t, subclasses included. Passing a class name string creates a forward reference that is resolved when first needed, allowing specs for classes not yet defined.

Examples:

# A required instance of class A and its subclasses.
pg.typing.Object(A)

# A required instance of class A and its subclasses (forward declaration).
pg.typing.Object('A')

# An optional instance of class A with None as its default value.
pg.typing.Object(A).noneable()

# An instance of class A with default value.
pg.typing.Object(A, default=A())

Constructor.

Parameters:

Name Type Description Default
t type[Any] | ForwardRef | str

Class of the object. Objects of subclass of this class is acceptable.

required
default Any

(Optional) default value of this spec.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
transform None | Callable[[Any], Any]

(Optional) user-defined function to be called on the input of apply. It could be used as a type converter or a custom validator which may raise errors.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    t: type[typing.Any] | class_schema.ForwardRef | str,
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    transform: None | (typing.Callable[[typing.Any], typing.Any]) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      t: Class of the object. Objects of subclass of this class is acceptable.
      default: (Optional) default value of this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      transform: (Optional) user-defined function to be called on the input
        of `apply`. It could be used as a type converter or a custom
        validator which may raise errors.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    if t is None:
        raise TypeError('"cls" for Object spec cannot be None.')

    forward_ref = None
    type_args = []
    if isinstance(t, class_schema.ForwardRef):
        forward_ref = t
    elif isinstance(t, str):
        forward_ref = class_schema.ForwardRef(
            _get_spec_callsite_module(), t
        )
    elif isinstance(t, type):
        if t is object:
            raise TypeError(
                "<class 'object'> is too general for Object spec."
            )
    elif getattr(t, '__no_type_check__', False):
        t = object
    elif not pg_inspect.is_generic(t):
        raise TypeError(
            f'"cls" for Object spec should be a type or str. Encountered: {t!r}.'
        )

    self._forward_ref = forward_ref
    self._type_args = type_args
    super().__init__(
        t,  # pyright: ignore[reportArgumentType]
        default,
        default_factory=default_factory,
        transform=transform,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

forward_refs property

forward_refs: set[ForwardRef]

Returns forward references used in this spec.

cls property

cls: type[Any]

Returns the class of this object spec.

schema property

schema: Schema | None

Returns the schema of object class if available.

extend

extend(base: ValueSpec) -> Self

Extend current value spec on top of a base spec.

Source code in pygx/typing/_value_specs.py
def extend(self, base: ValueSpec) -> typing.Self:
    """Extend current value spec on top of a base spec."""
    if isinstance(base, Callable) and base.is_compatible(self):
        return self
    return super().extend(base)

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    if self._forward_ref is not None:
        name = self._forward_ref.name
    else:
        # A parameterized spec renders subscripted (`Object(A[int])`,
        # #594) — the shared display helper, so the native renderer
        # (which calls it over the bridge) cannot drift from this one.
        name = pg_inspect.type_display_name(self._value_type)
    return formatting.kvlist_str(
        [
            ('', formatting.RawText(name), None),
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Optional

Optional(*args, **kwargs)

Bases: GenericTypeAlias

Type alias: Optional[T] is the spec of T marked noneable.

spec = pg.typing.Optional[int]
assert spec == pg.typing.Int().noneable()
Source code in pygx/typing/_value_specs.py
def __init__(self, *args, **kwargs):
    del args, kwargs
    raise TypeError(
        f'Generic type alias {self.__class__.__name__} cannot be instantiated.'
    )

PrimitiveType

PrimitiveType(
    value_type: type[Any] | tuple[type[Any], ...] | None,
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: ValueSpecBase

Base class of value specification for primitive types.

Constructor.

Parameters:

Name Type Description Default
value_type type[Any] | tuple[type[Any], ...] | None

Acceptable value type(s).

required
default Any

Default value.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    value_type: (type[typing.Any] | tuple[type[typing.Any], ...] | None),
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):
    """Constructor.

    Args:
      value_type: Acceptable value type(s).
      default: Default value.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    super().__init__(
        value_type,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Sequence

Sequence(*args, **kwargs)

Bases: GenericTypeAlias

Type alias: Sequence[T] accepts a list OR a tuple of T.

spec = pg.typing.Sequence[int]
assert spec == pg.typing.Union(
    [pg.typing.List(pg.typing.Int()), pg.typing.Tuple(pg.typing.Int())]
)
Source code in pygx/typing/_value_specs.py
def __init__(self, *args, **kwargs):
    del args, kwargs
    raise TypeError(
        f'Generic type alias {self.__class__.__name__} cannot be instantiated.'
    )

Str

Str(
    default: str | None = MISSING_VALUE,
    default_factory: Callable[[], str | None] | None = None,
    regex: str | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, PrimitiveType

Value spec for string type.

Accepts str values. An optional regex is validated with re.match — anchored at the start of the string only, so end the pattern with $ to require a full match.

Examples:

# A required str value.
pg.typing.Str()

# A required str value which matches with a regular expression.
pg.typing.Str(regex='foo.*')

# A str value with the default value set to 'foo'.
pg.typing.Str(default='foo')

# An optional str value with default value set to None.
pg.typing.Str().noneable()

# An optional str value with default value set to 'foo'.
pg.typing.Str(default='foo').noneable()

# A frozen str with value set to 'foo' that is not modifiable by subclasses.
pg.typing.Str().freeze('foo')

Constructor.

Parameters:

Name Type Description Default
default str | None

Default value for this value spec.

MISSING_VALUE
default_factory Callable[[], str | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
regex str | None

Optional regular expression for acceptable value, matched with re.match (anchored at the start only).

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    default: str | None = MISSING_VALUE,
    default_factory: typing.Callable[[], str | None] | None = None,
    regex: str | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      default: Default value for this value spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      regex: Optional regular expression for acceptable value, matched
        with ``re.match`` (anchored at the start only).
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    self._regex = re.compile(regex) if regex else None
    super().__init__(
        str,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )
    # Str._validate only does work when a regex is set.
    self._validate_is_noop = self._regex is None
    self._invalidate_fast_apply_info()  # `_validate_is_noop` finalized.

regex property

regex

Returns regular expression for acceptable values.

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    regex_pattern = self._regex.pattern if self._regex else None
    return formatting.kvlist_str(
        [
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('regex', regex_pattern, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Tuple

Tuple(
    element_values: ValueSpecOrAnnotation | Sequence[ValueSpecOrAnnotation],
    default: tuple[Any, ...] | None = MISSING_VALUE,
    default_factory: Callable[[], tuple[Any, ...] | None] | None = None,
    min_size: int | None = None,
    max_size: int | None = None,
    size: int | None = None,
    transform: None | Callable[[Any], tuple[Any, ...]] = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    *,
    min_length: int | None = None,
    max_length: int | None = None
)

Bases: Generic, ValueSpecBase

Value spec for tuple type.

Accepts tuple values, in two flavors. The fixed-length form (a sequence of element specs) validates each position against its own spec and requires an exact length; the variable-length form (a single element spec) validates every element against it, with optional min_size / max_size bounds.

Examples:

# A required tuple with 2 items which are non-negative integers.
pg.typing.Tuple([pg.typing.Int(min_value=0), pg.typing.Int(min_value=0)])

# An optional int tuple of size 3 with None as its default value.
pg.typing.Tuple(pg.typing.Int(), size=3).noneable()

# A int tuple with minimal size 1 and maximal size 5.
pg.typing.Tuple(pg.typing.Int(), min_size=1, max_size=5)

# A (int, float) tuple with default value (1, 1.0).
pg.typing.Tuple([pg.typing.Int(), pg.typing.Float()], default=(1, 1.0))

# A frozen tuple that prevents subclass to extend/override.
pg.typing.Tuple(pg.typing.Int()).freeze((1,))

Constructor.

Parameters:

Name Type Description Default
element_values ValueSpecOrAnnotation | Sequence[ValueSpecOrAnnotation]

A ValueSpec object or its equivalence as the element spec for a variable-length tuple, or a sequence of ValueSpec objects or their equivalences as the elements specs for a fixed-length tuple.

required
default tuple[Any, ...] | None

(Optional) default value for this spec.

MISSING_VALUE
default_factory Callable[[], tuple[Any, ...] | None] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
min_size int | None

(Optional) min size of tuple. If None, 0 will be used. Applicable only for variable-length tuple.

None
max_size int | None

(Optional) max size of tuple. Applicable only for variable-length tuple.

None
size int | None

(Optional) size of tuple. A shortcut to specify min_size and max_size at the same time. size and min_size/max_size are mutually exclusive.

None
transform None | Callable[[Any], tuple[Any, ...]]

(Optional) user-defined function to be called on the input of apply. It could be used as a type converter or a custom validator which may raise errors.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
min_length int | None

Synonym of min_size (#522, pydantic spelling). Mutually exclusive with it. Keyword-only.

None
max_length int | None

Synonym of max_size. Mutually exclusive with it. Keyword-only.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    element_values: (
        ValueSpecOrAnnotation | typing.Sequence[ValueSpecOrAnnotation]
    ),
    default: tuple[typing.Any, ...] | None = MISSING_VALUE,
    default_factory: typing.Callable[[], tuple[typing.Any, ...] | None]
    | None = None,
    min_size: int | None = None,
    max_size: int | None = None,
    size: int | None = None,
    transform: None
    | (typing.Callable[[typing.Any], tuple[typing.Any, ...]]) = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
    *,
    min_length: int | None = None,
    max_length: int | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      element_values: A ValueSpec object or its equivalence as the element spec
        for a variable-length tuple, or a sequence of ValueSpec objects or their
        equivalences as the elements specs for a fixed-length tuple.
      default: (Optional) default value for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      min_size: (Optional) min size of tuple. If None, 0 will be used.
        Applicable only for variable-length tuple.
      max_size: (Optional) max size of tuple. Applicable only for
        variable-length tuple.
      size: (Optional) size of tuple. A shortcut to specify min_size and
        max_size at the same time. `size` and `min_size`/`max_size` are
        mutually exclusive.
      transform: (Optional) user-defined function to be called on the input
        of `apply`. It could be used as a type converter or a custom
        validator which may raise errors.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
      min_length: Synonym of ``min_size`` (#522, pydantic spelling).
        Mutually exclusive with it. Keyword-only.
      max_length: Synonym of ``max_size``. Mutually exclusive with it.
        Keyword-only.
    """
    min_size, max_size = _normalize_size_aliases(
        'Tuple', min_size, max_size, min_length, max_length
    )
    if isinstance(element_values, (tuple, list)):
        if not element_values:
            raise ValueError(
                "Argument 'element_values' must be a non-empty list."
            )
        if size is not None or min_size is not None or max_size is not None:
            raise ValueError(
                '"size", "min_size" and "max_size" are not applicable '
                f'for fixed-length Tuple with elements: {element_values!r}'
            )
        element_values = [
            ValueSpec.from_annotation(v, auto_typing=True)
            for v in element_values
        ]
        min_size = len(element_values)
        max_size = min_size
    else:
        element_values = ValueSpec.from_annotation(
            element_values, auto_typing=True
        )
        if size is not None and (
            min_size is not None or max_size is not None
        ):
            raise ValueError(
                f'Either "size" or "min_size"/"max_size" pair can be specified. '
                f'Encountered: size={size}, min_size={min_size}, '
                f'max_size={max_size}.'
            )
        if size is not None:
            min_size = size
            max_size = size
        if min_size is None:
            min_size = 0
        if min_size < 0:
            raise ValueError(
                f'"min_size" of List must be no less than 0. '
                f'Encountered: {min_size}.'
            )
        if max_size is not None:
            if max_size < min_size:
                raise ValueError(
                    f'"max_size" of List must be no less than "min_size". '
                    f'Encountered: min_size={min_size}, max_size=max_size.'
                )
        if min_size == max_size:
            element_values = [element_values] * min_size

    if isinstance(element_values, ValueSpec):
        elements = [
            Field(
                key_specs.TupleKey(None),
                element_values,
                'Field of variable-length tuple element',
            )
        ]
    else:
        elements = [
            Field(
                key_specs.TupleKey(i),
                element_value,
                f'Field of tuple element at {i}',
            )
            for i, element_value in enumerate(element_values)
        ]

    self._min_size = min_size
    self._max_size = max_size
    self._elements = elements
    super().__init__(
        tuple,
        default,
        default_factory=default_factory,
        transform=transform,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )
    # Precomputed bulk-check state for `apply()`.
    #   _fixed_bulk_check: tuple of expected types per index (fixed-length
    #     tuples where every element spec is fast-eligible).
    #   _var_bulk_type_singleton: `frozenset({elem_type})` for the
    #     `set(map(type, value)) <= singleton` idiom (variable-length).
    # Either being non-None enables a `apply()` mega-fast path that
    # returns the input unchanged when every element already matches.
    self._fixed_bulk_check = None
    self._var_bulk_type = None
    self._var_bulk_type_singleton = None
    if min_size == max_size:
        types: list[typing.Any] | None = []
        for e in elements:
            ev = typing.cast(ValueSpecBase, e.value)
            ev_info = ev.fast_apply_info
            if not ev_info.is_eligible:
                types = None
                break
            types.append(ev_info.value_type)
        if types is not None:
            self._fixed_bulk_check = tuple(types)
    else:
        # Variable-length Tuple always carries exactly one element spec
        # (see constructor: a single ValueSpec is wrapped in a 1-element
        # Field list before reaching here).
        elem0 = typing.cast(ValueSpecBase, elements[0].value)
        elem0_info = elem0.fast_apply_info
        if elem0_info.is_eligible:
            self._var_bulk_type = elem0_info.value_type
            self._var_bulk_type_singleton = frozenset(
                (elem0_info.value_type,)
            )

fixed_length property

fixed_length: bool

Returns True if current Tuple spec is fixed length.

min_size property

min_size: int

Returns min size of this tuple.

max_size property

max_size: int | None

Returns max size of this tuple.

elements property

elements: list[Field]

Returns Field specification for tuple elements.

forward_refs

forward_refs() -> set[ForwardRef]

Returns forward references used in this spec.

Source code in pygx/typing/_value_specs.py
@cached_property
@override
def forward_refs(self) -> set[class_schema.ForwardRef]:
    """Returns forward references used in this spec."""
    forward_refs = set()
    for elem in self.elements:
        forward_refs.update(elem.value.forward_refs)
    return forward_refs

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    if self.fixed_length:
        value = [f.value for f in self._elements]
        default_min, default_max = self._min_size, self._max_size
    else:
        value = self._elements[0].value
        default_min, default_max = 0, None
    return formatting.kvlist_str(
        [
            ('', value, None),
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('min_size', self._min_size, default_min),
            ('max_size', self._max_size, default_max),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Type

Type(
    t: type[Any] | str,
    default: type[Any] = MISSING_VALUE,
    default_factory: Callable[[], type[Any]] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, ValueSpecBase

Value spec for type.

Accepts type objects that are t itself or one of its subclasses. Like Object, a class name string creates a forward reference resolved when first needed.

Examples:

# A required type or subclass of A.
pg.typing.Type(A)

# A required type or subclass of A (forward declaration).
pg.typing.Type('A')

# An optional type or subclass of A.
pg.typing.Type(A).noneable()

# A required type or subclass of A with default value B
# (B is a subclass of A).
pg.typing.Type(A, default=B)

Constructor.

Parameters:

Name Type Description Default
t type[Any] | str

The base type, or a string as a forward reference to it. Applied values must be t or one of its subclasses.

required
default type[Any]

(Optional) default type for this spec.

MISSING_VALUE
default_factory Callable[[], type[Any]] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
is_noneable bool

If True, None is acceptable.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    t: type[typing.Any] | str,
    default: type[typing.Any] = MISSING_VALUE,
    default_factory: typing.Callable[[], type[typing.Any]] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      t: The base type, or a string as a forward reference to it.
        Applied values must be ``t`` or one of its subclasses.
      default: (Optional) default type for this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting.
    """
    forward_ref = None
    if isinstance(t, str):
        forward_ref = class_schema.ForwardRef(
            _get_spec_callsite_module(), t
        )
    elif not (
        isinstance(t, type) or pg_inspect.is_generic(t) or t is typing.Any
    ):
        raise TypeError(f'{t!r} is not a type.')
    self._expected_type = t
    self._forward_ref = forward_ref
    super().__init__(
        type,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

type property

type: type[Any]

Returns desired type.

forward_refs

forward_refs() -> set[ForwardRef]

Returns forward references used in this spec.

Source code in pygx/typing/_value_specs.py
@cached_property
@override
def forward_refs(self) -> set[class_schema.ForwardRef]:
    """Returns forward references used in this spec."""
    if self._forward_ref is None:
        return set()
    return {self._forward_ref}

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('', self._expected_type, None),
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        **kwargs,
    )

Union

Union(
    candidates: Sequence[ValueSpecOrAnnotation],
    default: Any = MISSING_VALUE,
    default_factory: Callable[[], Any] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
)

Bases: Generic, ValueSpecBase

Value spec for Union.

A value is accepted when one of the candidate specs accepts it. Candidates are matched in declaration order: a value whose type matches a candidate is validated by that candidate; otherwise each candidate in turn is asked to coerce the value through its own apply (e.g. an int widens through a Float candidate), and the first success wins. If no candidate accepts, TypeError is raised. A union may not contain two candidates of the same spec type for the same value type (e.g. two Int candidates); two Object candidates of different classes are fine.

Examples:

# A required int or float value.
pg.typing.Union([pg.typing.Int(), pg.typing.Float()])

# An optional int or float value with default set to None.
pg.typing.Union([pg.typing.Int(), pg.typing.Float()]).noneable()

# A dict of specific keys, instance of class A or B, with {x=1} as its
# default value.
pg.typing.Union([
    pg.typing.Dict([
        ('x', pg.typing.Int(min_value=1)),
    ]),
    pg.typing.Object(A),
    pg.typing.Object(B),
], default={'x': 1})

Values are dispatched to the matching candidate; a mismatched type is coerced by the first candidate that accepts it:

spec = pg.typing.Union([pg.typing.Str(), pg.typing.Float()])
assert spec.apply('a') == 'a'
assert spec.apply(1) == 1.0  # Widened by the Float candidate.

Constructor.

Parameters:

Name Type Description Default
candidates Sequence[ValueSpecOrAnnotation]

A sequence of value spec objects or their equivalence as the spec for candidate types.

required
default Any

(Optional) default value of this spec.

MISSING_VALUE
default_factory Callable[[], Any] | None

Zero-arg callable invoked once per default materialization. Mutually exclusive with default.

None
is_noneable bool

(Optional) If True, None is acceptable for this spec.

False
frozen bool

If True, values other than the default value are not acceptable.

False
strict bool | None

If True, reject implicit conversions (strict mode); None (default) inherits the class-wide strict setting. Each candidate honors its own strict bit during coercion.

None
Source code in pygx/typing/_value_specs.py
def __init__(
    self,
    candidates: typing.Sequence[ValueSpecOrAnnotation],
    default: typing.Any = MISSING_VALUE,
    default_factory: typing.Callable[[], typing.Any] | None = None,
    is_noneable: bool = False,
    frozen: bool = False,
    strict: bool | None = None,
):  # pytype: disable=annotation-type-mismatch
    """Constructor.

    Args:
      candidates: A sequence of value spec objects or their equivalence as the
        spec for candidate types.
      default: (Optional) default value of this spec.
      default_factory: Zero-arg callable invoked once per default
        materialization. Mutually exclusive with `default`.
      is_noneable: (Optional) If True, None is acceptable for this spec.
      frozen: If True, values other than the default value are not
        acceptable.
      strict: If True, reject implicit conversions (strict mode); None
        (default) inherits the class-wide ``strict`` setting. Each
        candidate honors its own ``strict`` bit during coercion.
    """
    if not isinstance(candidates, (tuple, list)) or len(candidates) < 2:
        raise ValueError(
            f"Argument 'candidates' must be a list of at least 2 "
            f'elements. Encountered {candidates}.'
        )
    candidates = [
        ValueSpec.from_annotation(c, auto_typing=True) for c in candidates
    ]
    candidates_by_type = {}
    has_noneable_candidate = False
    for c in candidates:
        if c.is_noneable:
            has_noneable_candidate = True

        # NOTE(daiyip): When forward declaration is present (e.g.
        # `pg.typing.Object('A')`), calling `value_type` will trigger the
        # resolution of the referred type. Therefore, we refer to `_value_type`
        # instead of `value_type` to access the class name string in such cases.
        spec_type = (c.__class__, getattr(c, '_value_type'))
        if spec_type not in candidates_by_type:
            candidates_by_type[spec_type] = []
        candidates_by_type[spec_type].append(c)

    for spec_type, cs in candidates_by_type.items():
        if len(cs) > 1:
            # NOTE(daiyip): Now we simply reject union of multiple value spec of
            # the same type. We may consider support Union of different List, Tuple,
            # Dict and Object later.
            raise ValueError(
                f'Found {len(cs)} value specs of the same type {spec_type}.'
            )

    # CANDIDATE order, deduped through a dict (insertion-ordered) —
    # never a set (#596): the tuple is byte-visible in `Expect (...)`
    # errors, and set iteration order is process-dependent (type-object
    # hashes are address-based). The native ctor builds the same
    # ordered tuple (`union.rs`).
    candidate_types: dict[typing.Any, None] = {}
    no_value_type_check = False
    for c in candidates:
        child_value_type = getattr(c, '_value_type')
        if child_value_type is None:
            no_value_type_check = True
        elif isinstance(child_value_type, tuple):
            for t in child_value_type:
                candidate_types[t] = None
        else:
            candidate_types[child_value_type] = None

    self._candidates = candidates
    union_value_type = (
        None if no_value_type_check else tuple(candidate_types)
    )
    super().__init__(
        union_value_type,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable or has_noneable_candidate,
        frozen=frozen,
        strict=strict,
    )
    # `type(value) -> candidate` dispatch map for `_apply` and `apply`.
    # Covers the common `Union[int, str, ...]` case in O(1); falls back to
    # the linear `isinstance` scan when any candidate has a non-plain
    # value_type (tuple, generic, None, forward-ref) or when two
    # candidates claim the same type.
    self._candidate_type_map = self._build_candidate_type_map(
        candidates, no_value_type_check
    )

candidates property

candidates: list[ValueSpec]

Returns candidate types of this union spec.

forward_refs

forward_refs() -> set[ForwardRef]

Returns forward references used in this spec.

Source code in pygx/typing/_value_specs.py
@cached_property
@override
def forward_refs(self) -> set[class_schema.ForwardRef]:
    """Returns forward references used in this spec."""
    forward_refs = set()
    for c in self._candidates:
        forward_refs.update(c.forward_refs)
    return forward_refs

noneable

noneable(is_noneable: bool = True, use_none_as_default: bool = True) -> Self

Union-specific noneable: propagates to every candidate.

Besides marking the union itself, each candidate spec is marked noneable (or not) as well, so None is acceptable no matter which candidate handles the value.

Source code in pygx/typing/_value_specs.py
def noneable(
    self, is_noneable: bool = True, use_none_as_default: bool = True
) -> Self:
    """Union-specific noneable: propagates to every candidate.

    Besides marking the union itself, each candidate spec is marked
    noneable (or not) as well, so ``None`` is acceptable no matter
    which candidate handles the value.
    """
    super().noneable(
        is_noneable=is_noneable, use_none_as_default=use_none_as_default
    )
    for c in self._candidates:
        c.noneable(is_noneable=is_noneable, use_none_as_default=False)
    return self

get_candidate

get_candidate(dest_spec: ValueSpec) -> ValueSpec | None

Get candidate by a destination value spec.

Parameters:

Name Type Description Default
dest_spec ValueSpec

destination value spec which is a superset of the value spec to return. E.g. Any (dest_spec) is superset of Int (child spec).

required

Returns:

Type Description
ValueSpec | None

The first value spec under Union with which the destination value spec is compatible.

Source code in pygx/typing/_value_specs.py
def get_candidate(self, dest_spec: ValueSpec) -> ValueSpec | None:
    """Get candidate by a destination value spec.

    Args:
      dest_spec: destination value spec which is a superset of the value spec
        to return. E.g. Any (dest_spec) is superset of Int (child spec).

    Returns:
      The first value spec under Union with which the destination value spec
        is compatible.
    """
    # NOTE(daiyip): we always try matching the candidate with the same
    # value spec type first, then different type but compatible specs.
    for c in self._candidates:
        if dest_spec.__class__ == c.__class__ and dest_spec.is_compatible(
            c
        ):
            return c

    for c in self._candidates:
        if isinstance(c, Union):
            child = c.get_candidate(dest_spec)
            if child is not None:
                return child
        else:
            if dest_spec.is_compatible(c):
                return c
    return None

is_compatible

is_compatible(other: ValueSpec) -> bool

Union specific compatibility check.

Source code in pygx/typing/_value_specs.py
def is_compatible(self, other: ValueSpec) -> bool:
    """Union specific compatibility check."""
    if isinstance(other, Union):
        for oc in other.candidates:
            if not self.is_compatible(oc):
                return False
        return True
    else:
        for c in self._candidates:
            if c.is_compatible(other):
                return True
        return False

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/typing/_value_specs.py
def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    return formatting.kvlist_str(
        [
            ('', self._candidates, None),
            ('default', self._default, MISSING_VALUE),
            ('default_factory', self._default_factory, None),
            ('noneable', self._is_noneable, False),
            ('frozen', self._frozen, False),
            ('strict', self._strict, None),
        ],
        label=self.__class__.__name__,
        compact=compact,
        verbose=verbose,
        root_indent=root_indent,
        list_wrap_threshold=kwargs.pop('list_wrap_threshold', 20),
        **kwargs,
    )

CallableWithOptionalKeywordArgs

CallableWithOptionalKeywordArgs(
    func: Callable[..., Any], optional_keywords: list[str]
)

Wraps a callable, dropping keyword args it does not accept.

Calls delegate to func, with any keyword listed in optional_keywords that is absent from func's signature silently removed. Useful for invoking user callbacks that may or may not declare an optional keyword.

Example:

f = pg.typing.CallableWithOptionalKeywordArgs(lambda x: x ** 2, ['y'])
assert f(2, y=3) == 4  # Keyword 'y' is ignored.
Source code in pygx/typing/_callable_ext.py
def __init__(self, func: Callable[..., Any], optional_keywords: list[str]):
    sig = callable_signature.signature(
        func, auto_typing=False, auto_doc=False
    )

    # Check for variable keyword arguments.
    if sig.has_varkw:
        absent_keywords = []
    else:
        all_keywords = set(sig.arg_names)
        absent_keywords = [
            k for k in optional_keywords if k not in all_keywords
        ]
    self._absent_keywords = absent_keywords
    self._func = func

PresetArgValue

PresetArgValue(default: Any = MISSING_VALUE)

Bases: Formattable

Marker default for arguments whose values may come from presets.

Used as an argument's default value in a function decorated with enable_preset_args: when the caller omits the argument, its value is taken from the enclosing preset_args context, falling back to default when no preset provides it.

Example:

@pg.typing.enable_preset_args()
def foo(x, y=pg.typing.PresetArgValue(default=1)):
  return x + y

with pg.typing.preset_args(dict(y=2)):
  print(foo(x=1))  # 3: y=2
print(foo(x=1))    # 2: y=1
Source code in pygx/typing/_callable_ext.py
def __init__(self, default: Any = topology.MISSING_VALUE):
    self.default = default

inspect classmethod

inspect(func: FunctionType) -> dict[str, PresetArgValue]

Gets the PresetArgValue specified in a function's signature.

Source code in pygx/typing/_callable_ext.py
@classmethod
def inspect(cls, func: types.FunctionType) -> dict[str, 'PresetArgValue']:
    """Gets the PresetArgValue specified in a function's signature."""
    assert inspect.isfunction(func), func
    sig = inspect.signature(func)
    preset_arg_markers = {}
    for p in sig.parameters.values():
        if isinstance(p.default, cls):
            preset_arg_markers[p.name] = p.default
    return preset_arg_markers

resolve_args classmethod

resolve_args(
    call_args: Sequence[Any],
    call_kwargs: dict[str, Any],
    positional_arg_names: Sequence[str],
    arg_defaults: dict[str, Any],
    preset_kwargs: dict[str, Any],
    include_all_preset_kwargs: bool = False,
) -> tuple[Sequence[Any], dict[str, Any]]

Resolves calling arguments passed to a method with presets.

Source code in pygx/typing/_callable_ext.py
@classmethod
def resolve_args(
    cls,
    call_args: Sequence[Any],
    call_kwargs: dict[str, Any],
    positional_arg_names: Sequence[str],
    arg_defaults: dict[str, Any],
    preset_kwargs: dict[str, Any],
    include_all_preset_kwargs: bool = False,
) -> tuple[Sequence[Any], dict[str, Any]]:
    """Resolves calling arguments passed to a method with presets."""
    # Step 1: compute marked kwargs.
    resolved_kwargs = {}
    for arg_name, arg_default in arg_defaults.items():
        if not isinstance(arg_default, PresetArgValue):
            resolved_kwargs[arg_name] = arg_default
            continue
        if arg_name in preset_kwargs:
            resolved_kwargs[arg_name] = preset_kwargs[arg_name]
        elif arg_default.has_default:
            resolved_kwargs[arg_name] = arg_default.default
        else:
            raise ValueError(
                f'Argument {arg_name!r} is not present as a keyword argument '
                'from the caller.'
            )

    # Step 2: add preset kwargs
    if include_all_preset_kwargs:
        for k, v in preset_kwargs.items():
            if k not in resolved_kwargs:
                resolved_kwargs[k] = v

    # Step 3: merge call kwargs with resolved preset kwargs.
    resolved_kwargs.update(call_kwargs)

    # Step 3: remove resolved kwargs items as it's present in call args.
    for i in range(len(call_args)):
        if i >= len(positional_arg_names):
            break
        resolved_kwargs.pop(positional_arg_names[i], None)

    # Step 4: convert kwargs back to postional arguments if applicable
    resolved_args = call_args
    if len(positional_arg_names) > len(call_args):
        resolved_args = list(resolved_args)
        for arg_name in positional_arg_names[len(call_args) :]:
            if arg_name not in resolved_kwargs:
                break
            arg_value = resolved_kwargs.pop(arg_name)
            resolved_args.append(arg_value)
    return resolved_args, resolved_kwargs

cached_property

cached_property(func: Callable[[Any], _T])

Bases: property, Generic[_T]

Property-typed view of functools.cached_property.

Pyright sees this as a property[_T] subclass so overrides of @property-typed base attributes type-check. At runtime the bound name is functools.cached_property (see the else branch); this class is never instantiated.

Source code in pygx/typing/_typing_compat.py
def __init__(self, func: Callable[[Any], _T]) -> None:  # pylint: disable=unused-argument
    ...

annotation_from_str

annotation_from_str(
    annotation_str: str, parent_module: ModuleType | None = None
) -> Any

Parses a type annotation from its string form.

Example:

assert pg.typing.annotation_from_str('list[str]') == list[str]

BNF for PyType annotations:

<maybe_union>      ::= <type> | <type> "|" <maybe_union>
<type>             ::= <literal_type> | <non_literal_type>

<literal_type>     ::= "Literal" <literal_params>
<literal_params>   ::= "[" <python_values> "]"
                       (parsed by `pg.coding.evaluate`)

<non_literal_type> ::= <type_id> | <type_id> "[" <type_arg> "]"
<type_arg>         ::= <maybe_type_list>
                       | <maybe_type_list> "," <maybe_type_list>
<maybe_type_list>  ::= "[" <type_arg> "]" | <maybe_union>
<type_id>          ::= 'aAz_.1-9'

Parameters:

Name Type Description Default
annotation_str str

String form of type annotations. E.g. "list[str]".

required
parent_module ModuleType | None

The module where the annotation was defined, used to resolve non-builtin names.

None

Returns:

Type Description
Any

Object form of the annotation.

Raises:

Type Description
SyntaxError

If the annotation string is invalid.

Source code in pygx/typing/_annotation_conversion.py
def annotation_from_str(
    annotation_str: str,
    parent_module: types.ModuleType | None = None,
) -> typing.Any:
    """Parses a type annotation from its string form.

    **Example:**

    ```python
    assert pg.typing.annotation_from_str('list[str]') == list[str]
    ```

    BNF for PyType annotations:

    ```text
    <maybe_union>      ::= <type> | <type> "|" <maybe_union>
    <type>             ::= <literal_type> | <non_literal_type>

    <literal_type>     ::= "Literal" <literal_params>
    <literal_params>   ::= "[" <python_values> "]"
                           (parsed by `pg.coding.evaluate`)

    <non_literal_type> ::= <type_id> | <type_id> "[" <type_arg> "]"
    <type_arg>         ::= <maybe_type_list>
                           | <maybe_type_list> "," <maybe_type_list>
    <maybe_type_list>  ::= "[" <type_arg> "]" | <maybe_union>
    <type_id>          ::= 'aAz_.1-9'
    ```

    Args:
      annotation_str: String form of type annotations. E.g. "list[str]".
      parent_module: The module where the annotation was defined, used to
        resolve non-builtin names.

    Returns:
      Object form of the annotation.

    Raises:
      SyntaxError: If the annotation string is invalid.
    """
    s = annotation_str
    context = dict(pos=0)

    def _eof() -> bool:
        return context['pos'] == len(s)

    def _pos() -> int:
        return context['pos']

    def _next(n: int = 1, offset: int = 0) -> str:
        if _eof():
            return '<EOF>'
        return s[_pos() + offset : _pos() + offset + n]

    def _advance(n: int) -> None:
        context['pos'] += n

    def _error_illustration() -> str:
        return f'{s}\n{" " * _pos()}' + '^'

    def _match(ch) -> bool:
        if _next(len(ch)) == ch:
            _advance(len(ch))
            return True
        return False

    def _skip_whitespaces() -> None:
        while _next() in ' \t':
            _advance(1)

    def _maybe_union():
        t = _type()
        while not _eof():
            _skip_whitespaces()
            if _match('|'):
                t = t | _type()  # pyright: ignore[reportOperatorIssue]
            else:
                break
        return t

    def _type():
        type_id = _type_id()
        t = _resolve(type_id)
        if t is typing.Literal:
            return t[_literal_params()]
        elif _match('['):
            arg = _type_arg()
            if not _match(']'):
                raise SyntaxError(
                    f'Expected "]" at position {_pos()}.\n\n'
                    + _error_illustration()
                )
            return t[arg]  # pyright: ignore[reportIndexIssue,reportOptionalSubscript]
        return t

    def _literal_params():
        if not _match('['):
            raise SyntaxError(
                f'Expected "[" at position {_pos()}.\n\n'
                + _error_illustration()
            )
        arg_start = _pos()
        in_str = False
        escape_mode = False
        num_open_bracket = 1

        while num_open_bracket > 0:
            ch = _next()
            if _eof():
                raise SyntaxError(
                    f'Unexpected end of annotation at position {_pos()}.\n\n'
                    + _error_illustration()
                )
            if ch == '\\':
                escape_mode = not escape_mode
            else:
                escape_mode = False

            if ch == "'" and not escape_mode:
                in_str = not in_str
            elif not in_str:
                if ch == '[':
                    num_open_bracket += 1
                elif ch == ']':
                    num_open_bracket -= 1
            _advance(1)

        arg_str = s[arg_start : _pos() - 1]
        return coding.evaluate(
            '(' + arg_str + ')', permission=coding.CodePermission.BASIC
        )

    def _type_arg():
        t_args = []
        t_args.append(_maybe_type_list())
        while _match(','):
            t_args.append(_maybe_type_list())
        return tuple(t_args) if len(t_args) > 1 else t_args[0]

    def _maybe_type_list():
        if _match('['):
            ret = _type_arg()
            if not _match(']'):
                raise SyntaxError(
                    f'Expected "]" at position {_pos()}.\n\n'
                    + _error_illustration()
                )
            return list(ret) if isinstance(ret, tuple) else [ret]
        return _maybe_union()

    def _type_id() -> str:
        _skip_whitespaces()
        if _match('...'):
            return '...'
        start = _pos()
        while not _eof():
            c = _next()
            if c.isalnum() or c in '_.':
                _advance(1)
            else:
                break
        t_id = s[start : _pos()]
        if not all(x.isidentifier() for x in t_id.split('.')):
            raise SyntaxError(
                f'Expected type identifier, got {t_id!r} at position {start}.\n\n'
                + _error_illustration()
            )
        return t_id

    def _resolve(type_id: str):

        def _as_forward_ref() -> typing.ForwardRef:
            if sys.version_info >= (3, 14):  # pragma: no cover
                return typing.ForwardRef(
                    type_id
                )  # pytype: disable=not-callable
            return typing.ForwardRef(
                type_id, False, parent_module
            )  # pytype: disable=not-callable

        def _resolve_name(name: str, parent_obj: typing.Any):
            if name == 'None':
                return None, True
            if parent_obj is not None and hasattr(parent_obj, name):
                return getattr(parent_obj, name), False
            if hasattr(builtins, name):
                return getattr(builtins, name), True
            if type_id == '...':
                return ..., True
            return topology.MISSING_VALUE, False

        names = type_id.split('.')
        if len(names) == 1:
            reference, is_builtin = _resolve_name(names[0], parent_module)
            if is_builtin:
                return reference
            if not is_builtin and (
                # When reference is not found, we should treat it as a forward
                # reference.
                reference == topology.MISSING_VALUE
                # When module is being reloaded, we should treat all non-builtin
                # references as forward references.
                or getattr(parent_module, '__reloading__', False)
            ):
                return _as_forward_ref()
            return reference

        root_obj, _ = _resolve_name(names[0], parent_module)
        # When root object is not found, we should treat it as a forward reference.
        if root_obj == topology.MISSING_VALUE:
            return _as_forward_ref()

        parent_obj = root_obj
        # When root object is a module, we should treat reference to its children
        # as non-forward references.
        if inspect.ismodule(root_obj):
            for name in names[1:]:
                parent_obj, _ = _resolve_name(name, parent_obj)
                if parent_obj == topology.MISSING_VALUE:
                    raise TypeError(f'{type_id!r} does not exist.')
            return parent_obj
        # When root object is non-module variable of current module, and when the
        # module is being reloaded, we should treat reference to its children as
        # forward references.
        elif getattr(parent_module, '__reloading__', False):
            return _as_forward_ref()
        # When root object is non-module variable of current module, we should treat
        # unresolved reference to its children as forward references.
        else:
            for name in names[1:]:
                parent_obj, _ = _resolve_name(name, parent_obj)
                if parent_obj == topology.MISSING_VALUE:
                    return _as_forward_ref()
            return parent_obj

    root = _maybe_union()
    if _pos() != len(s):
        raise SyntaxError(
            'Unexpected end of annotation.\n\n' + _error_illustration()
        )
    return root

validate

validate(
    value: Any,
    annotation: Any,
    *,
    allow_partial: bool = False,
    parent_module: ModuleType | None = None
) -> Any

Validates a bare value against a type annotation, in one line.

The TypeAdapter analog for pydantic migrants: converts annotation into a value spec (cached per annotation) and applies it to value — type-checking, running registered coercions (e.g. int → float), and filling schema defaults — without hand-building the spec:

pg.validate([1, 2], list[int])            # -> [1, 2]
pg.validate({'x': 1}, dict[str, int])     # -> {'x': 1}
pg.validate(None, int | None)             # -> None
pg.validate([1, 'x'], list[int])          # TypeError (path=[1])

Parameters:

Name Type Description Default
value Any

The value to validate. Never mutated — containers are validated on a copy (inplace=False), so default-filling returns a new container and leaves the input untouched.

required
annotation Any

A type annotation (list[int], dict[str, int] | None, a pg.Object subclass, a string forward reference, ...) or an existing ValueSpec, which is applied as-is (uncached).

required
allow_partial bool

If True, missing required values are filled with MISSING_VALUE placeholders instead of raising.

False
parent_module ModuleType | None

The module whose namespace resolves string forward references in annotation. Required for string annotations naming module-local types; resolution is cached per (annotation, module) pair at first call.

None

Returns:

Type Description
Any

The validated (possibly coerced / default-filled) value.

Raises:

Type Description
TypeError

The value (or a nested element, path-prefixed) does not match the annotation — or the annotation contains a forward reference that does not resolve at call time (a typo'd string, or a missing parent_module); class-field annotations defer such references, a one-shot validation call must not.

ValueError

A value-level constraint fails (e.g. None for a non-nullable spec, missing required key).

KeyError

A mapping value carries a key its schema does not allow (the pg.Dict-idiom error map, via a passed-through spec).

Source code in pygx/typing/_annotation_conversion.py
def validate(
    value: typing.Any,
    annotation: typing.Any,
    *,
    allow_partial: bool = False,
    parent_module: types.ModuleType | None = None,
) -> typing.Any:
    """Validates a bare value against a type annotation, in one line.

    The `TypeAdapter` analog for pydantic migrants: converts ``annotation``
    into a value spec (cached per annotation) and applies it to ``value`` —
    type-checking, running registered coercions (e.g. ``int`` → ``float``),
    and filling schema defaults — without hand-building the spec:

    ```python
    pg.validate([1, 2], list[int])            # -> [1, 2]
    pg.validate({'x': 1}, dict[str, int])     # -> {'x': 1}
    pg.validate(None, int | None)             # -> None
    pg.validate([1, 'x'], list[int])          # TypeError (path=[1])
    ```

    Args:
      value: The value to validate. Never mutated — containers are validated
        on a copy (``inplace=False``), so default-filling returns a new
        container and leaves the input untouched.
      annotation: A type annotation (``list[int]``, ``dict[str, int] | None``,
        a ``pg.Object`` subclass, a string forward reference, ...) or an
        existing ``ValueSpec``, which is applied as-is (uncached).
      allow_partial: If True, missing required values are filled with
        ``MISSING_VALUE`` placeholders instead of raising.
      parent_module: The module whose namespace resolves string forward
        references in ``annotation``. Required for string annotations naming
        module-local types; resolution is cached per ``(annotation, module)``
        pair at first call.

    Returns:
      The validated (possibly coerced / default-filled) value.

    Raises:
      TypeError: The value (or a nested element, path-prefixed) does not
        match the annotation — or the annotation contains a forward
        reference that does not resolve at call time (a typo'd string, or
        a missing ``parent_module``); class-field annotations defer such
        references, a one-shot validation call must not.
      ValueError: A value-level constraint fails (e.g. ``None`` for a
        non-nullable spec, missing required key).
      KeyError: A mapping value carries a key its schema does not allow
        (the ``pg.Dict``-idiom error map, via a passed-through spec).
    """
    if isinstance(annotation, class_schema.ValueSpec):
        spec = annotation
    else:
        key = (annotation, parent_module)
        try:
            spec = _VALIDATE_SPEC_CACHE.get(key)
        except TypeError:
            # Unhashable annotation (e.g. carrying an unhashable metadata
            # object) — convert uncached.
            spec = None
            key = None
        if spec is None:
            spec = _value_spec_from_annotation(
                annotation, auto_typing=True, parent_module=parent_module
            )
            if (
                key is not None
                # An `Annotated[...]` may carry the USER's own spec by
                # reference (`pg.field(value_spec=...)` metadata); a later
                # class definition mutates it via `set_default`, so a cached
                # entry would become order-dependent (#533 review m1).
                and not hasattr(annotation, '__metadata__')
                # Forward refs with no explicit module bind to the CALLER's
                # module (stack inspection) — caching would serve the first
                # caller's namespace to every module (#533 review B1).
                and not (parent_module is None and spec.forward_refs)
            ):
                _VALIDATE_SPEC_CACHE[key] = spec
        # A forward ref unresolved at CALL time runs no type check — a
        # typo'd string annotation would silently accept everything. Class
        # fields legitimately defer (recursive classes); a one-shot
        # validation call must surface it (#533 review M3, owner ruling).
        unresolved = [r.name for r in spec.forward_refs if not r.resolved]
        if unresolved:
            raise TypeError(
                f'pg.validate: annotation {annotation!r} contains forward '
                f'reference(s) {unresolved} that do not resolve. Check the '
                f'spelling, define the referenced type(s) first, or pass '
                f'`parent_module=` naming the module that defines them.'
            )
    return spec.apply(value, allow_partial=allow_partial, inplace=False)

schema

schema(
    cls_or_fn: Callable[..., Any],
    args: (
        list[Field | FieldDef] | dict[FieldKeyDef, FieldValueDef] | None
    ) = None,
    returns: ValueSpec | None = None,
    *,
    auto_typing: bool = True,
    auto_doc: bool = True,
    remove_self: bool = True,
    include_return: bool = False
) -> Schema

Returns the schema from the signature of a class or a function.

Parameters:

Name Type Description Default
cls_or_fn Callable[..., Any]

A class or a function.

required
args list[Field | FieldDef] | dict[FieldKeyDef, FieldValueDef] | None

(Optional) additional annotations for arguments.

None
returns ValueSpec | None

(Optional) additional annotation for return value.

None
auto_typing bool

If True, enable type inference from annotations.

True
auto_doc bool

If True, extract schema/field descriptions from docstrings.

True
remove_self bool

If True, remove the first self argument if it appears in the signature.

True
include_return bool

If True, include the return value spec in the schema under the key 'return'.

False

Returns:

Type Description
Schema

A pg.typing.Schema object.

Source code in pygx/typing/_callable_signature.py
def schema(
    cls_or_fn: Callable[..., Any],
    args: (
        list[class_schema.Field | class_schema.FieldDef]
        | dict[class_schema.FieldKeyDef, class_schema.FieldValueDef]
        | None
    ) = None,
    returns: class_schema.ValueSpec | None = None,
    *,
    auto_typing: bool = True,
    auto_doc: bool = True,
    remove_self: bool = True,
    include_return: bool = False,
) -> class_schema.Schema:
    """Returns the schema from the signature of a class or a function.

    Args:
      cls_or_fn: A class or a function.
      args: (Optional) additional annotations for arguments.
      returns: (Optional) additional annotation for return value.
      auto_typing: If True, enable type inference from annotations.
      auto_doc: If True, extract schema/field descriptions from docstrings.
      remove_self: If True, remove the first `self` argument if it appears in the
        signature.
      include_return: If True, include the return value spec in the schema
        under the key 'return'.

    Returns:
      A pg.typing.Schema object.
    """
    s = getattr(cls_or_fn, '__schema__', None)
    if isinstance(s, class_schema.Schema):
        return s
    return (
        signature(cls_or_fn, auto_typing=auto_typing, auto_doc=auto_doc)
        .annotate(args, return_value=returns)
        .to_schema(
            remove_self=remove_self,
            include_return=include_return,
        )
    )

signature

signature(
    func: Callable[..., Any], auto_typing: bool = True, auto_doc: bool = True
) -> Signature

Gets the signature from a Python callable.

Source code in pygx/typing/_callable_signature.py
def signature(
    func: Callable[..., Any],
    auto_typing: bool = True,
    auto_doc: bool = True,
) -> Signature:  # pylint:disable=g-bare-generic
    """Gets the signature from a Python callable."""
    return Signature.from_callable(func, auto_typing, auto_doc)

copy_default

copy_default(value: Any, memo: dict[int, Any] | None = None) -> Any

Selective per-instance copy of a static default (Pydantic-aligned, #664).

The rule (owner ruling 2026-07-28, matching Pydantic v2 as probed on 2.13.4): deep-copy the recognized copyable shapes — mutable builtin containers (deeply, opaque elements included), symbolic values (via their own copy protocol; topology.MaybePartial is the marker), dataclass instances, and types that opt in via topology.CopyIfDefault — and SHARE everything else by reference: the arbitrary-type template instance is reused across constructs, which also makes unpicklable defaults (locks, client handles) legal. Incidental deepcopyability is NOT the gate — an arbitrary object defining __deepcopy__ is still shared, exactly like Pydantic; extending CopyIfDefault is the declared per-type intent, and default_factory remains the per-field escape hatch.

memo threads an enclosing __deepcopy__'s memo into the copy arms (ValueSpecBase.__deepcopy__ routes its _default through here so a spec clone keeps the shared template shared). The SHARE arm is memo-coherent: it consults the memo first (another slot of the same deepcopy may already have placed this object) and registers the shared template otherwise — so sibling slots referencing the same object (an Enum's _values candidate that IS the default) alias to ONE object on the clone, preserving intra-spec invariants like default in values.

Source code in pygx/typing/_class_schema.py
def copy_default(value: Any, memo: dict[int, Any] | None = None) -> Any:
    """Selective per-instance copy of a static default (Pydantic-aligned, #664).

    The rule (owner ruling 2026-07-28, matching Pydantic v2 as probed on
    2.13.4): deep-copy the *recognized copyable shapes* — mutable builtin
    containers (deeply, opaque elements included), symbolic values (via
    their own copy protocol; `topology.MaybePartial` is the marker),
    dataclass instances, and types that opt in via
    `topology.CopyIfDefault` — and SHARE everything else by reference:
    the arbitrary-type template instance is reused across constructs,
    which also makes unpicklable defaults (locks, client handles) legal.
    Incidental deepcopyability is NOT the gate — an arbitrary object
    defining `__deepcopy__` is still shared, exactly like Pydantic;
    extending `CopyIfDefault` is the *declared* per-type intent, and
    `default_factory` remains the per-field escape hatch.

    ``memo`` threads an enclosing `__deepcopy__`'s memo into the copy arms
    (`ValueSpecBase.__deepcopy__` routes its `_default` through here so a
    spec clone keeps the shared template shared). The SHARE arm is
    memo-coherent: it consults the memo first (another slot of the same
    deepcopy may already have placed this object) and registers the
    shared template otherwise — so sibling slots referencing the same
    object (an `Enum`'s `_values` candidate that IS the default) alias to
    ONE object on the clone, preserving intra-spec invariants like
    `default in values`.
    """
    if type(value) in _ATOMIC_DEFAULT_TYPES:
        return value
    if isinstance(value, _COPYABLE_CONTAINER_TYPES):
        return copy.deepcopy(value, memo)
    if isinstance(value, (topology.MaybePartial, topology.CopyIfDefault)):
        return copy.deepcopy(value, memo)
    if dataclasses.is_dataclass(value) and not isinstance(value, type):
        return copy.deepcopy(value, memo)
    if memo is not None:
        vid = id(value)
        existing = memo.get(vid)
        if existing is not None:
            return existing
        memo[vid] = value
    return value

create_field

create_field(
    field_or_def: Field | FieldDef,
    auto_typing: bool = True,
    accept_value_as_annotation: bool = True,
    parent_module: ModuleType | None = None,
) -> Field

Creates Field from its equivalence.

Parameters:

Name Type Description Default
field_or_def Field | FieldDef

a Field object or its equivalence, which is a tuple of 2 - 4 elements: (<key>, <value>, [description], [metadata]). key can be a KeySpec subclass object or string. value can be a ValueSpec subclass object or equivalent value. (see ValueSpec.from_annotation method). description is the description of this field. It can be optional when this field overrides the default value of a field defined in parent schema. metadata is an optional field which is a dict of user objects.

required
auto_typing bool

If True, infer value spec from Python annotations. Otherwise, pg.typing.Any() will be used.

True
accept_value_as_annotation bool

If True, allow default values to be used as annotations when creating the value spec.

True
parent_module ModuleType | None

(Optional) parent module for defining this field, which will be used for forward reference lookup.

None

Returns:

Type Description
Field

A Field object.

Source code in pygx/typing/_class_schema.py
def create_field(
    field_or_def: Field | FieldDef,
    auto_typing: bool = True,
    accept_value_as_annotation: bool = True,
    parent_module: types.ModuleType | None = None,
) -> Field:
    """Creates ``Field`` from its equivalence.

    Args:
      field_or_def: a ``Field`` object or its equivalence, which is a tuple of
        2 - 4 elements:
        `(<key>, <value>, [description], [metadata])`.
        `key` can be a KeySpec subclass object or string. `value` can be a
        ValueSpec subclass object or equivalent value. (see
        ``ValueSpec.from_annotation`` method). `description` is the
        description of this field. It can be optional when this field
        overrides the default value of a field defined in parent schema.
        `metadata` is an optional field which is a dict of user objects.
      auto_typing: If True, infer value spec from Python annotations. Otherwise,
        ``pg.typing.Any()`` will be used.
      accept_value_as_annotation: If True, allow default values to be used as
        annotations when creating the value spec.
      parent_module: (Optional) parent module for defining this field, which will
        be used for forward reference lookup.

    Returns:
      A ``Field`` object.
    """
    if isinstance(field_or_def, Field):
        return field_or_def

    if not isinstance(field_or_def, tuple):
        raise TypeError(
            f'Field definition should be tuples with 2 to 4 elements. '
            f'Encountered: {field_or_def}.'
        )

    field_def = list(field_or_def)
    if len(field_def) == 4:
        maybe_key_spec, maybe_value_spec, description, field_metadata = (
            field_def
        )
    elif len(field_def) == 3:
        maybe_key_spec, maybe_value_spec, description = field_def
        field_metadata = {}
    elif len(field_def) == 2:
        maybe_key_spec, maybe_value_spec = field_def
        description = None
        field_metadata = {}
    else:
        raise TypeError(
            f'Field definition should be tuples with 2 to 4 elements. '
            f'Encountered: {field_or_def}.'
        )

    if isinstance(maybe_key_spec, (str, KeySpec)):
        key = maybe_key_spec
    else:
        raise TypeError(
            f'The 1st element of field definition should be of '
            f"<class 'str'> or KeySpec. Encountered: {maybe_key_spec}."
        )

    value = ValueSpec.from_annotation(
        maybe_value_spec,
        auto_typing=auto_typing,
        accept_value_as_annotation=accept_value_as_annotation,
        parent_module=parent_module,
    )

    if description is not None and not isinstance(description, str):
        raise TypeError(
            f'Description (the 3rd element) of field definition '
            f'should be text type. Encountered: {description}'
        )
    if not isinstance(field_metadata, dict):
        raise TypeError(
            f'Metadata (the 4th element) of field definition '
            f'should be a dict of objects. '
            f'Encountered: {field_metadata}'
        )
    field = Field(key, value, description, field_metadata)
    # NOTE (§6): the former `_annotation_forces_wrap` seam lived here — a
    # subscripted `pg.Dict[str, V]` / `pg.List[E]` annotation used to force
    # `symbolic=True` on the field SPEC. The annotation now authors
    # `symbolic=True` on the SPEC itself (at every depth, not just the
    # field's top level), so the field-level flag is redundant and gone.
    return field

create_schema

create_schema(
    fields: dict[FieldKeyDef, FieldValueDef] | list[Field | FieldDef],
    name: str | None = None,
    base_schema_list: list[Schema] | None = None,
    allow_nonconst_keys: bool = False,
    metadata: dict[str, Any] | None = None,
    description: str | None = None,
    for_cls: type[Any] | None = None,
    parent_module: ModuleType | None = None,
) -> Schema

Creates Schema from a list of Fields or equivalences.

Examples:

The following code examples are equivalent:

schema = pg.typing.create_schema({
    'a': int,
    'b': (int, 'field b'),
    'c': (pg.typing.Str(regex='.*'), 'field c', dict(meta1=1))
})

schema = pg.typing.create_schema([
    ('a', int),
    ('b', int, 'field b'),
    ('c', pg.typing.Str(regex='.*'), 'field c', dict(meta1=1)),
])

schema = pg.typing.create_schema([
    pg.typing.Field('a', pg.typing.Int()),
    pg.typing.Field('b', pg.typing.Int(), 'field b'),
    pg.typing.Field(
        'c', pg.typing.Str(regex='.*'), 'field c', dict(meta1=1)),
])

Parameters:

Name Type Description Default
fields dict[FieldKeyDef, FieldValueDef] | list[Field | FieldDef]

A dict, a list of field or equivalent values. A Field equivalent value is either a Field object or a tuple of 2 - 4 elements: (<key>, <value>, [description], [metadata]). key can be a KeySpec subclass object or string. value can be a ValueSpec subclass object or equivalent value. (see ValueSpec.from_annotation method). description is the description of this field. It can be optional when this field overrides the default value of a field defined in parent schema. metadata is an optional field which is a dict of user objects.

required
name str | None

An optional name for the schema.

None
base_schema_list list[Schema] | None

A list of schema objects as bases.

None
allow_nonconst_keys bool

Whether to allow non const keys in schema.

False
metadata dict[str, Any] | None

Optional dict of user objects as schema-level metadata.

None
description str | None

Optional description of the schema.

None
for_cls type[Any] | None

(Optional) the class that this schema applies to.

None
parent_module ModuleType | None

(Optional) parent module for defining this schema, which will be used for forward reference lookup.

None

Returns:

Type Description
Schema

Schema object.

Raises:

Type Description
TypeError

If input type is incorrect.

Source code in pygx/typing/_class_schema.py
def create_schema(
    fields: (
        dict[FieldKeyDef, FieldValueDef] | list[Field | FieldDef]  # pylint: disable=g-bare-generic
    ),
    name: str | None = None,
    base_schema_list: list[Schema] | None = None,
    allow_nonconst_keys: bool = False,
    metadata: dict[str, Any] | None = None,
    description: str | None = None,
    for_cls: type[Any] | None = None,
    parent_module: types.ModuleType | None = None,
) -> Schema:
    """Creates ``Schema`` from a list of ``Field``s or equivalences.

    Examples:

      The following code examples are equivalent:

      ```python
      schema = pg.typing.create_schema({
          'a': int,
          'b': (int, 'field b'),
          'c': (pg.typing.Str(regex='.*'), 'field c', dict(meta1=1))
      })

      schema = pg.typing.create_schema([
          ('a', int),
          ('b', int, 'field b'),
          ('c', pg.typing.Str(regex='.*'), 'field c', dict(meta1=1)),
      ])

      schema = pg.typing.create_schema([
          pg.typing.Field('a', pg.typing.Int()),
          pg.typing.Field('b', pg.typing.Int(), 'field b'),
          pg.typing.Field(
              'c', pg.typing.Str(regex='.*'), 'field c', dict(meta1=1)),
      ])
      ```

    Args:
      fields: A dict, a list of field or equivalent values. A Field equivalent
        value is either a Field object or a tuple of 2 - 4 elements: `(<key>,
        <value>, [description], [metadata])`. `key` can be a KeySpec subclass
        object or string. `value` can be a ValueSpec subclass object or equivalent
        value. (see ``ValueSpec.from_annotation`` method). `description` is the
        description of this field. It can be optional when this field overrides
        the default value of a field defined in parent schema. `metadata` is an
        optional field which is a dict of user objects.
      name: An optional name for the schema.
      base_schema_list: A list of schema objects as bases.
      allow_nonconst_keys: Whether to allow non const keys in schema.
      metadata: Optional dict of user objects as schema-level metadata.
      description: Optional description of the schema.
      for_cls: (Optional) the class that this schema applies to.
      parent_module: (Optional) parent module for defining this schema, which will
        be used for forward reference lookup.

    Returns:
      Schema object.

    Raises:
      TypeError: If input type is incorrect.
    """
    fields = _normalize_field_defs(fields)
    metadata = metadata or {}
    if not isinstance(metadata, dict):
        raise TypeError(
            f'Metadata of schema should be a dict. Encountered: {metadata}.'
        )

    return Schema(
        fields=[
            create_field(field_or_def, parent_module=parent_module)
            for field_or_def in fields
        ],
        name=name,
        base_schema_list=base_schema_list,
        allow_nonconst_keys=allow_nonconst_keys,
        metadata=metadata,
        description=description,
        for_cls=for_cls,
    )

docstr

docstr(symbol: Any) -> DocStr | None

Gets the structured docstring of a Python symbol, or None if absent.

Source code in pygx/typing/_docstr.py
def docstr(symbol: Any) -> DocStr | None:
    """Gets the structured docstring of a Python symbol, or None if absent."""
    docstr_text = getattr(symbol, '__doc__', None)
    return DocStr.parse(docstr_text) if docstr_text else None

callable_eq

callable_eq(x: Any, y: Any) -> bool

Returns True if two (maybe) callables are equal.

For functions, x and y are considered equal when they are the same instance or have the same code (e.g. two identical lambda x: x).

For methods, x and y are considered equal when:

  • Static method: the same method from the same class hierarchy, e.g. a subclass inheriting a base class' static method.
  • Class method: the same method from the same class. An inherited class method is considered a different class method.
  • Instance method: when self is not bound, the same method from the same class hierarchy (like static methods); when self is bound, the same method on the same object.

Parameters:

Name Type Description Default
x Any

An optional function or method object.

required
y Any

An optional function or method object.

required

Returns:

Type Description
bool

True if x and y are considered equal — either the same instance,

bool

or derived from the same code with the same effect.

Source code in pygx/typing/_inspection.py
def callable_eq(x: Any, y: Any) -> bool:
    """Returns True if two (maybe) callables are equal.

    For functions, `x` and `y` are considered equal when they are the same
    instance or have the same code (e.g. two identical ``lambda x: x``).

    For methods, `x` and `y` are considered equal when:

      * Static method: the same method from the same class hierarchy, e.g.
        a subclass inheriting a base class' static method.
      * Class method: the same method from the same class. An inherited
        class method is considered a different class method.
      * Instance method: when `self` is not bound, the same method from the
        same class hierarchy (like static methods); when `self` is bound,
        the same method on the same object.

    Args:
      x: An optional function or method object.
      y: An optional function or method object.

    Returns:
      True if `x` and `y` are considered equal — either the same instance,
      or derived from the same code with the same effect.
    """
    if x is y:
        return True
    if x is None or y is None:
        return False
    if inspect.isfunction(x) and inspect.isfunction(y):
        return _code_eq(x.__code__, y.__code__)
    elif inspect.ismethod(x) and inspect.ismethod(y):
        return (
            _code_eq(x.__code__, y.__code__) and x.__self__ is y.__self__
        )  # pytype: disable=attribute-error
    return x == y

get_outer_class

get_outer_class(
    cls: type[Any],
    base_cls: type[Any] | tuple[type[Any], ...] | None = None,
    immediate: bool = False,
) -> type[Any] | None

Returns the class enclosing cls in its nesting hierarchy.

Resolution follows cls.__qualname__ through the module's globals, walking the nesting chain from the innermost enclosing class outwards.

Example:

class A1:
  class B:
    class C:
      ...

assert pg.typing.get_outer_class(A1.B) is A1
assert pg.typing.get_outer_class(A1.B.C) is A1.B
# With `base_cls`, the walk returns the innermost enclosing class
# that subclasses it (A1.B is not an A1 subclass; A1 subclasses
# itself).
assert pg.typing.get_outer_class(A1.B.C, base_cls=A1) is A1
assert pg.typing.get_outer_class(A1.B.C, base_cls=int) is None
assert pg.typing.get_outer_class(A1) is None

Parameters:

Name Type Description Default
cls type[Any]

The class to get the outer class for.

required
base_cls type[Any] | tuple[type[Any], ...] | None

The base class the returned outer class must subclass. If provided, enclosing classes that are not base_cls subclasses are skipped (or, with immediate=True, cause None to be returned).

None
immediate bool

Whether to consider only the immediate outer class, which is returned as None if it is not a subclass of base_cls. Applicable when base_cls is not None.

False

Returns:

Type Description
type[Any] | None

The outer class of cls. None if there is none, or no enclosing

type[Any] | None

class satisfies base_cls.

Raises:

Type Description
ValueError

cls is defined locally inside a function (its outer class cannot be resolved from the qualified name).

Source code in pygx/typing/_inspection.py
def get_outer_class(
    cls: type[Any],
    base_cls: type[Any] | tuple[type[Any], ...] | None = None,
    immediate: bool = False,
) -> type[Any] | None:
    """Returns the class enclosing ``cls`` in its nesting hierarchy.

    Resolution follows ``cls.__qualname__`` through the module's globals,
    walking the nesting chain from the innermost enclosing class outwards.

    **Example:**

    ```python
    class A1:
      class B:
        class C:
          ...

    assert pg.typing.get_outer_class(A1.B) is A1
    assert pg.typing.get_outer_class(A1.B.C) is A1.B
    # With `base_cls`, the walk returns the innermost enclosing class
    # that subclasses it (A1.B is not an A1 subclass; A1 subclasses
    # itself).
    assert pg.typing.get_outer_class(A1.B.C, base_cls=A1) is A1
    assert pg.typing.get_outer_class(A1.B.C, base_cls=int) is None
    assert pg.typing.get_outer_class(A1) is None
    ```

    Args:
      cls: The class to get the outer class for.
      base_cls: The base class the returned outer class must subclass. If
        provided, enclosing classes that are not `base_cls` subclasses are
        skipped (or, with `immediate=True`, cause None to be returned).
      immediate: Whether to consider only the immediate outer class, which
        is returned as None if it is not a subclass of `base_cls`.
        Applicable when `base_cls` is not None.

    Returns:
      The outer class of `cls`. None if there is none, or no enclosing
      class satisfies `base_cls`.

    Raises:
      ValueError: `cls` is defined locally inside a function (its outer
        class cannot be resolved from the qualified name).
    """
    if '<locals>' in cls.__qualname__:
        raise ValueError(
            'Cannot find the outer class for locally defined class '
            f'{cls.__qualname__!r}'
        )

    names = cls.__qualname__.split('.')
    if len(names) < 2:
        return None

    parent = sys.modules[cls.__module__]
    symbols = []
    for name in names[:-1]:
        symbol = getattr(parent, name, None)
        if symbol is None:
            return None
        assert inspect.isclass(symbol), symbol
        symbols.append(symbol)
        parent = symbol

    for symbol in reversed(symbols):
        if immediate:
            return (
                symbol if not base_cls or issubclass(symbol, base_cls) else None
            )
        if not base_cls or issubclass(symbol, base_cls):
            return symbol
    return None

get_type

get_type(maybe_type: Any) -> type[Any]

Gets the type of a maybe generic type.

Source code in pygx/typing/_inspection.py
def get_type(maybe_type: Any) -> type[Any]:
    """Gets the type of a maybe generic type."""
    if isinstance(maybe_type, type):
        return maybe_type
    origin = typing.get_origin(maybe_type)
    if origin is not None:
        return origin
    else:
        raise TypeError(f'{maybe_type!r} is not a type.')

get_type_args

get_type_args(
    maybe_generic: type[Any], base: type[Any] | None = None
) -> tuple[type[Any], ...]

Gets generic type args conditioned on an optional base class.

With base, resolves the args maybe_generic provides for that generic base through the WHOLE inheritance chain, substituting TypeVars hop by hop (#592) — class P(A[int]); class Q(P, Generic[Y]) resolves (int,) for A, and class W(B[X], Generic[X]); W[str] resolves (str,) for B. Returns () when the chain never binds the base's params (the gradual-arm signal in is_subclass, #590); an unbound TypeVar surviving substitution stays in the result (the arg loop erases it).

Source code in pygx/typing/_inspection.py
def get_type_args(
    maybe_generic: type[Any], base: type[Any] | None = None
) -> tuple[type[Any], ...]:
    """Gets generic type args conditioned on an optional base class.

    With ``base``, resolves the args ``maybe_generic`` provides for that
    generic base through the WHOLE inheritance chain, substituting TypeVars
    hop by hop (#592) — ``class P(A[int]); class Q(P, Generic[Y])`` resolves
    ``(int,)`` for ``A``, and ``class W(B[X], Generic[X]); W[str]`` resolves
    ``(str,)`` for ``B``. Returns ``()`` when the chain never binds the
    base's params (the gradual-arm signal in `is_subclass`, #590); an
    unbound TypeVar surviving substitution stays in the result (the arg
    loop erases it).
    """
    if base is None:
        return typing.get_args(maybe_generic)
    return _resolved_type_args(maybe_generic, base, {})

has_generic_bases

has_generic_bases(maybe_generic: Any) -> bool

Returns True if a type is a generic subclass.

Source code in pygx/typing/_inspection.py
def has_generic_bases(maybe_generic: Any) -> bool:
    """Returns True if a type is a generic subclass."""
    return bool(getattr(maybe_generic, '__orig_bases__', None))

is_generic

is_generic(maybe_generic: Any) -> bool

Returns True if a type is a generic class.

Source code in pygx/typing/_inspection.py
def is_generic(maybe_generic: Any) -> bool:
    """Returns True if a type is a generic class."""
    return typing.get_origin(maybe_generic) is not None

is_instance

is_instance(value: Any, target: Any) -> bool

An isinstance extension that supports Any and generic types.

Equivalent to is_subclass(type(value), target) — see is_subclass for the compatibility rules.

Source code in pygx/typing/_inspection.py
def is_instance(value: Any, target: Any) -> bool:
    """An ``isinstance`` extension that supports ``Any`` and generic types.

    Equivalent to ``is_subclass(type(value), target)`` — see `is_subclass`
    for the compatibility rules.
    """
    return is_subclass(type(value), target)

is_subclass

is_subclass(src: Any, target: Any) -> bool

An issubclass extension that supports Any and generic types.

typing.Any as the target accepts every source. A subscripted generic target checks type args recursively, resolving the args the source binds for the target's generic base through its inheritance chain; unbound type parameters erase to Any (gradual semantics). target may also be a tuple of types, matching any member.

Example:

import typing

assert pg.typing.is_subclass(bool, int)
assert pg.typing.is_subclass(int, typing.Any)
assert pg.typing.is_subclass(list[int], typing.Sequence[int])
assert pg.typing.is_subclass(list, list[int])  # Unbound args ~ Any.
assert not pg.typing.is_subclass(list[str], list[int])
Source code in pygx/typing/_inspection.py
def is_subclass(src: Any, target: Any) -> bool:
    """An ``issubclass`` extension that supports ``Any`` and generic types.

    ``typing.Any`` as the target accepts every source. A subscripted
    generic target checks type args recursively, resolving the args the
    source binds for the target's generic base through its inheritance
    chain; unbound type parameters erase to ``Any`` (gradual semantics).
    ``target`` may also be a tuple of types, matching any member.

    **Example:**

    ```python
    import typing

    assert pg.typing.is_subclass(bool, int)
    assert pg.typing.is_subclass(int, typing.Any)
    assert pg.typing.is_subclass(list[int], typing.Sequence[int])
    assert pg.typing.is_subclass(list, list[int])  # Unbound args ~ Any.
    assert not pg.typing.is_subclass(list[str], list[int])
    ```
    """

    def _is_subclass(src: type[Any], target: type[Any]) -> bool:
        # Union support (#773 ruling 4): a union TARGET matches when any
        # member does; a union SOURCE is a subclass only when every member
        # is. Both spellings — `typing.Union[X, Y]` (which `issubclass`
        # rejects with a TypeError) and `X | Y` (which silently compared
        # False) — are handled.
        if typing.get_origin(target) in (typing.Union, types.UnionType):
            return any(_is_subclass(src, t) for t in typing.get_args(target))
        if typing.get_origin(src) in (typing.Union, types.UnionType):
            return all(_is_subclass(s, target) for s in typing.get_args(src))
        if is_protocol(target):
            # NOTE(daiyip): loose runtime check for Protocol.
            # As runtime protocol check (through decorating the protocol class with
            # @typing.runtime_checkable) might be unreliable and very slow.
            return inspect.isclass(src) and src.__module__ != 'builtins'
        if target is Any:
            return True
        elif src is Any:
            return False

        orig_target = typing.get_origin(target)
        orig_src = typing.get_origin(src)

        if orig_target is None:
            if orig_src is None:
                # Neither source nor target is a generic class.
                return issubclass(src, target)
            # Source class is generic but not the target class.
            return issubclass(orig_src, target)
        elif orig_src is None:
            # Target class is generic but not the source class.
            if not issubclass(src, orig_target):
                return False
        elif not issubclass(orig_src, orig_target):
            # Both are generic, but the source is not a subclass of the target.
            return False

        # Check type args.
        t_args = get_type_args(target)
        if not t_args:
            return True

        s_args = get_type_args(src, base=orig_target)
        if s_args:
            assert len(s_args) == len(t_args), (s_args, t_args)
            for s_arg, t_arg in zip(s_args, t_args):
                # An unbound TypeVar surviving the substitution walk erases
                # to `Any` per gradual semantics (#592) — that ARG is
                # compatible either way, without flagging the whole source
                # as unbound (its other args still discriminate). ParamSpec
                # and TypeVarTuple survivors (the latter spelled
                # `Unpack[Ts]`) get the same erasure — they are never
                # substituted by the walk.
                if _is_type_var_like(s_arg) or _is_type_var_like(t_arg):
                    continue
                if not _is_subclass(s_arg, t_arg):
                    return False
            return True
        else:
            # The source provides NO type arguments for the target's generic
            # base — a plain `A` against `A[int]`, a `class B1(B)` that never
            # binds `B`'s params, or a builtin-generic subclass like
            # `class MyDict(pg.Dict)` against `MyDict[str, int]`. Gradual
            # semantics (#590): unbound parameters erase to `Any`, and `Any`
            # is bidirectionally compatible — so the pair is compatible.
            # This is what makes `vs.Object(A[int])` accept direct `A`
            # instances (a runtime value's type never carries the
            # subscription). A source that binds the args ANYWHERE in its
            # inheritance chain never lands here — `get_type_args`
            # substitutes TypeVars hop by hop (#592), so
            # `class Q(P, Generic[Y])` over `class P(A[int])` resolves
            # `(int,)` and discriminates above. Remaining erasure boundary:
            # a TypeVar nested INSIDE an arg alias (`A[list[X]]`) is erased,
            # not substituted — the arg loop treats it as `Any`.
            return True

    if isinstance(target, tuple):
        return any(_is_subclass(src, x) for x in target)
    return _is_subclass(src, target)

substitute_type_params

substitute_type_params(hint: Any, env: dict[Any, Any]) -> Any

Substitutes type parameters in hint from env.

Top-level type variables directly, nested ones (list[T]) through alias re-subscription. Parameters absent from env erase to Any (gradual semantics, as is_subclass #590).

Source code in pygx/typing/_inspection.py
def substitute_type_params(hint: Any, env: dict[Any, Any]) -> Any:
    """Substitutes type parameters in ``hint`` from ``env``.

    Top-level type variables directly, nested ones (``list[T]``) through
    alias re-subscription. Parameters absent from ``env`` erase to ``Any``
    (gradual semantics, as `is_subclass` #590).
    """
    if isinstance(hint, typing.TypeVar):
        return env.get(hint, typing.Any)
    params = getattr(hint, '__parameters__', ())
    if params:
        return hint[tuple(env.get(p, typing.Any) for p in params)]
    return hint

ensure_value_spec

ensure_value_spec(
    value_spec: ValueSpec, src_spec: ValueSpec, root_path: KeyPath | None = None
) -> ValueSpec | None

Extracts the part of value_spec that matches src_spec.

For a Union destination, the candidate compatible with src_spec is selected; an Any destination imposes no constraint.

Parameters:

Name Type Description Default
value_spec ValueSpec

The destination value spec to match against (possibly a Union or Any).

required
src_spec ValueSpec

The source value spec whose values should be acceptable to the destination.

required
root_path KeyPath | None

An optional path for the value to include in error message.

None

Returns:

Type Description
ValueSpec | None

The matching (candidate) spec of value_spec, or None if value_spec is Any (no constraint to enforce).

Raises:

Type Description
TypeError

When src_spec is not compatible with value_spec.

Source code in pygx/typing/_value_specs.py
def ensure_value_spec(
    value_spec: class_schema.ValueSpec,
    src_spec: class_schema.ValueSpec,
    root_path: topology.KeyPath | None = None,
) -> class_schema.ValueSpec | None:
    """Extracts the part of ``value_spec`` that matches ``src_spec``.

    For a `Union` destination, the candidate compatible with ``src_spec``
    is selected; an `Any` destination imposes no constraint.

    Args:
      value_spec: The destination value spec to match against (possibly a
        `Union` or `Any`).
      src_spec: The source value spec whose values should be acceptable to
        the destination.
      root_path: An optional path for the value to include in error message.

    Returns:
      The matching (candidate) spec of ``value_spec``, or None if
        ``value_spec`` is `Any` (no constraint to enforce).

    Raises:
      TypeError: When ``src_spec`` is not compatible with ``value_spec``.
    """
    if isinstance(value_spec, Union):
        value_spec = value_spec.get_candidate(src_spec)  # pyright: ignore[reportAssignmentType]
    if isinstance(value_spec, Any):
        return None
    if not src_spec.is_compatible(value_spec):
        raise TypeError(
            topology.message_on_path(
                f'Source spec {src_spec} is not compatible with destination '
                f'spec {value_spec}.',
                root_path,
            )
        )
    return value_spec

enable_preset_args

enable_preset_args(
    include_all_preset_kwargs: bool = False, preset_name: str = "global"
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator factory for functions that may use preset argument values.

Arguments whose defaults are PresetArgValue markers pick up their values from the enclosing preset_args context at call time. Note the decorator must be called (@enable_preset_args(), with parentheses).

Example:

@pg.typing.enable_preset_args()
def foo(x, y=pg.typing.PresetArgValue(default=1)):
  return x + y

with pg.typing.preset_args(dict(y=2)):
  print(foo(x=1))  # 3: y=2
print(foo(x=1))    # 2: y=1

Parameters:

Name Type Description Default
include_all_preset_kwargs bool

Whether to also pass preset kwargs that are not marked as PresetArgValue when calling the function. Effective only when the function accepts **kwargs.

False
preset_name str

The name of the preset to read kwargs from.

'global'

Returns:

Type Description
Callable[[Callable[..., Any]], Callable[..., Any]]

A decorator that returns a function consuming preset argument values.

Callable[[Callable[..., Any]], Callable[..., Any]]

A function without PresetArgValue defaults is returned unchanged.

Source code in pygx/typing/_callable_ext.py
def enable_preset_args(
    include_all_preset_kwargs: bool = False, preset_name: str = 'global'
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator factory for functions that may use preset argument values.

    Arguments whose defaults are `PresetArgValue` markers pick up their
    values from the enclosing `preset_args` context at call time. Note the
    decorator must be called (``@enable_preset_args()``, with parentheses).

    **Example:**

    ```python
    @pg.typing.enable_preset_args()
    def foo(x, y=pg.typing.PresetArgValue(default=1)):
      return x + y

    with pg.typing.preset_args(dict(y=2)):
      print(foo(x=1))  # 3: y=2
    print(foo(x=1))    # 2: y=1
    ```

    Args:
      include_all_preset_kwargs: Whether to also pass preset kwargs that
        are not marked as `PresetArgValue` when calling the function.
        Effective only when the function accepts ``**kwargs``.
      preset_name: The name of the preset to read kwargs from.

    Returns:
      A decorator that returns a function consuming preset argument values.
      A function without `PresetArgValue` defaults is returned unchanged.
    """

    def decorator(func):
        sig = inspect.signature(func)
        positional_arg_names = [
            p.name
            for p in sig.parameters.values()
            if p.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD
        ]

        arg_defaults = {}
        has_preset_value = False
        has_varkw = False
        for p in sig.parameters.values():
            if p.kind == inspect.Parameter.VAR_KEYWORD:
                has_varkw = True
                continue
            if p.kind == inspect.Parameter.VAR_POSITIONAL:
                continue
            if p.default == inspect.Parameter.empty:
                continue
            if isinstance(p.default, PresetArgValue):
                has_preset_value = True
            arg_defaults[p.name] = p.default

        if has_preset_value:

            @functools.wraps(func)
            def _func(*args, **kwargs):
                # Map positional arguments to keyword arguments.
                presets = concurrent.thread_local_peek(
                    _TLS_KEY_PRESET_KWARGS, None
                )
                preset_kwargs = (
                    presets.get_preset(preset_name) if presets else {}
                )
                args, kwargs = PresetArgValue.resolve_args(
                    args,
                    kwargs,
                    positional_arg_names,
                    arg_defaults,
                    preset_kwargs,
                    include_all_preset_kwargs=include_all_preset_kwargs
                    and has_varkw,
                )
                return func(*args, **kwargs)

            return _func
        return func

    return decorator

preset_args

preset_args(
    kwargs: dict[str, Any],
    *,
    preset_name: str = "global",
    inherit_preset: str | bool = False
) -> Iterator[_ArgPresets]

Context manager providing preset kwargs to preset-enabled functions.

Within the context, functions decorated with enable_preset_args resolve their PresetArgValue-defaulted arguments from kwargs. See PresetArgValue for a usage example.

Parameters:

Name Type Description Default
kwargs dict[str, Any]

A dict of preset kwargs to be used by preset-enabled functions within the context.

required
preset_name str

The name of the preset to hold the kwargs. enable_preset_args allows users to pass a preset name, which will be used to identify the preset to read from.

'global'
inherit_preset str | bool

The name of the preset defined by the parent context to inherit kwargs from. Or a boolean to indicate whether to inherit a parent preset of the same name.

False

Yields:

Type Description
_ArgPresets

The current presets (an internal collection holding the preset

_ArgPresets

kwargs by preset name).

Source code in pygx/typing/_callable_ext.py
@contextlib.contextmanager
def preset_args(
    kwargs: dict[str, Any],
    *,
    preset_name: str = 'global',
    inherit_preset: str | bool = False,
) -> Iterator['_ArgPresets']:
    """Context manager providing preset kwargs to preset-enabled functions.

    Within the context, functions decorated with `enable_preset_args`
    resolve their `PresetArgValue`-defaulted arguments from ``kwargs``.
    See `PresetArgValue` for a usage example.

    Args:
      kwargs: A dict of preset kwargs to be used by preset-enabled
        functions within the context.
      preset_name: The name of the preset to hold the kwargs.
        `enable_preset_args` allows users to pass a preset name, which will
        be used to identify the preset to read from.
      inherit_preset: The name of the preset defined by the parent context
        to inherit kwargs from. Or a boolean to indicate whether to inherit
        a parent preset of the same name.

    Yields:
      The current presets (an internal collection holding the preset
      kwargs by preset name).
    """

    parent_presets = concurrent.thread_local_peek(
        _TLS_KEY_PRESET_KWARGS, _ArgPresets()
    )
    current_preset = parent_presets.derive(kwargs, preset_name, inherit_preset)
    concurrent.thread_local_push(_TLS_KEY_PRESET_KWARGS, current_preset)
    try:
        yield current_preset
    finally:
        concurrent.thread_local_pop(_TLS_KEY_PRESET_KWARGS, None)

to_json_schema

to_json_schema(
    value: ValueSpec | Schema | Any,
    *,
    include_type_name: bool = True,
    include_subclasses: bool = False,
    inline_nested_refs: bool = False
) -> dict[str, Any]

Converts a schema, value spec or type annotation to JSON schema.

Example:

class A(pg.Object):
  x: int = 1

assert pg.typing.to_json_schema(A, include_type_name=False) == {
    'type': 'object',
    'title': 'A',
    'properties': {'x': {'type': 'integer', 'default': 1}},
    'additionalProperties': False,
}

Parameters:

Name Type Description Default
value ValueSpec | Schema | Any

A pg.typing.Schema, a pg.typing.ValueSpec, or anything a value spec can be created from through annotation conversion (a type annotation, a pg.Object subclass, ...).

required
include_type_name bool

Whether class schemas include a _type const property carrying the PyGX type name, which allows the JSON values conforming to the schema to be loaded via pg.from_json.

True
include_subclasses bool

Whether an object spec emits an anyOf covering all non-abstract subclasses of its class.

False
inline_nested_refs bool

Whether, when a single-referenced definition is inlined, single-referenced definitions nested inside it are inlined too. Definitions referenced more than once always stay under $defs.

False

Returns:

Type Description
dict[str, Any]

A JSON schema dict.

Raises:

Type Description
TypeError

value (or a nested value spec) cannot be represented as JSON schema.

ValueError

An enum spec holds a candidate that has no JSON representation.

Source code in pygx/typing/_json_schema.py
def to_json_schema(
    value: vs.ValueSpec | class_schema.Schema | Any,
    *,
    include_type_name: bool = True,
    include_subclasses: bool = False,
    inline_nested_refs: bool = False,
) -> dict[str, Any]:
    """Converts a schema, value spec or type annotation to JSON schema.

    **Example:**

    ```python
    class A(pg.Object):
      x: int = 1

    assert pg.typing.to_json_schema(A, include_type_name=False) == {
        'type': 'object',
        'title': 'A',
        'properties': {'x': {'type': 'integer', 'default': 1}},
        'additionalProperties': False,
    }
    ```

    Args:
      value: A ``pg.typing.Schema``, a ``pg.typing.ValueSpec``, or anything
        a value spec can be created from through annotation conversion
        (a type annotation, a ``pg.Object`` subclass, ...).
      include_type_name: Whether class schemas include a ``_type`` const
        property carrying the PyGX type name, which allows the JSON values
        conforming to the schema to be loaded via ``pg.from_json``.
      include_subclasses: Whether an object spec emits an ``anyOf``
        covering all non-abstract subclasses of its class.
      inline_nested_refs: Whether, when a single-referenced definition is
        inlined, single-referenced definitions nested inside it are
        inlined too. Definitions referenced more than once always stay
        under ``$defs``.

    Returns:
      A JSON schema dict.

    Raises:
      TypeError: `value` (or a nested value spec) cannot be represented as
        JSON schema.
      ValueError: An enum spec holds a candidate that has no JSON
        representation.
    """
    defs = dict()
    if isinstance(value, class_schema.Schema):
        root = _json_schema_from_schema(
            value,
            defs=defs,
            type_name=value.name,
            include_type_name=include_type_name,
            include_subclasses=include_subclasses,
        )
    else:
        value = vs.ValueSpec.from_annotation(value, auto_typing=True)
        root = _json_schema_from_value_spec(
            value,
            defs=defs,
            include_type_name=include_type_name,
            include_subclasses=include_subclasses,
        )
    return _canonicalize_schema(
        root, defs, inline_nested_refs=inline_nested_refs
    )

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2