Skip to content

pygx.typing

Symbolic typing.

typing

Symbolic typing.

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 Runtime typing_ for more details.

  • Allow symbolic attributes to be placeheld by special symbols, in order to represent abstract concepts such as a space of objects. E.g. hyper primitives like pygx.oneof placeholds a symbolic attribute to create a space of values for that attribute. See Symbolic placeholding_ 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 difficut 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::

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=0.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=[])),
    ('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, default=[])),

    # 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([
        ('p1', pg.typing.Int()),
        ('p2', 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()],
                             kwargs=[('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))
])

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::

class A(pg.Object): x: 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: pg.typing.Int(max_value=5, default=2) z: pg.typing.Str('foo').freeze()

assert 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.

Symbolic placeholding

Symbolic placeholding enables scenarios that requires an abstract (non-material) representation of objects - for example - a search space or a partially bound object. Such objects with placeheld 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 = f(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 manipulated it: f.sym_rebind(x=1, y=2)

# Returns 3 f()

Symbolic placeholding is achieved by the pygx.CustomTyping interface, which was intended as a mechanism to extend the typing system horizontally without modifying the pygx.ValueSpec of existing classes' attributes.

This is done by allowing the CustomTyping subclasses to take over value validation and transformation from an attribute's original ValueSpec via pygx.CustomTyping.custom_apply method.

Following is an example of using CustomTyping to extend the schema system::

class FloatTensor(pg.typing.CustomTyping):

def __init__(self, tensor):
  self._tensor = tensor

def custom_apply(
    self, path, value_spec, allow_partial, child_transform):
  if isinstane(value_spec, pg.typing.Float):
    # Validate initial tensor, we can also add an TF operator to guard
    # tensor value to go beyond value_spec.min_value and max_value.
    value_spec.apply(self._tensor.numpy())
    # Shortcircuit .apply and returns object itself as final value.
    return (False, self)
  else:
    raise ValueError('FloatTensor can only be applied to float type.')

class Foo(pg.Object): x: float

# FloatTensor can be accepted for all symbolic attributes # with Float value spec. f = Foo(x=FloatTensor(tf.constant(1.0)))

MissingValue

Bases: Formattable, WireConvertible

Value placeholder for an unassigned attribute.

Argument dataclass 

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

Definition for a callable argument.

Kind

Bases: Enum

Arugment kind.

from_parameter classmethod 
from_parameter(parameter: Parameter) -> Kind

Returns Argument.Kind from inspect.Parameter.

Source code in pygx/typing/_callable_signature.py 
50
51
52
53
54
55
56
57
@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 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@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 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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

PY3 function signature.

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 (PY3).

 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 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
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 (PY3).
      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 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
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 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
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 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
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 
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
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 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
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 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
@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 
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
@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):
        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 
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
@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 
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
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,
    enable_wrap: bool | 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 if 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
enable_wrap bool | None

Whether raw dict / list values supplied to this field are wrapped into pg.Dict / pg.List at the field boundary. None resolves to enable_init at this (typing) layer — wrap for init fields, raw for non-init. At the pg.Object layer the sym axis governs the default: a sym=False class materializes enable_wrap=False onto its init fields, so a flat object stores raw containers. Strictly narrower than enable_validation: only gates the symbolic transform callback; type-validation and __pg_accept__-based coercions still run. Moot when enable_validation=False — the whole apply pipeline is skipped then, so the wrapping callback never fires.

 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 
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
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,
    enable_wrap: bool | 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__``.
      enable_wrap: Whether raw ``dict`` / ``list`` values
        supplied to this field are wrapped into ``pg.Dict`` /
        ``pg.List`` at the field boundary. ``None`` resolves to
        ``enable_init`` at this (typing) layer — wrap for init
        fields, raw for non-init. At the ``pg.Object`` layer the
        ``sym`` axis governs the default: a ``sym=False`` class
        materializes ``enable_wrap=False`` onto its init fields, so a
        flat object stores raw containers. Strictly narrower than
        ``enable_validation``: only gates the symbolic transform
        callback; type-validation and ``__pg_accept__``-based
        coercions still run. Moot when
        ``enable_validation=False`` — the whole apply pipeline is
        skipped then, so the wrapping callback never fires.
      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._enable_wrap = enable_wrap
    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 we can add multiple annotations to a field.

userdata = field.metadata.get('userdata', None):

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).

enable_wrap property 

enable_wrap: bool

Whether raw dict / list are wrapped into pg.Dict / pg.List.

Gates only the symbolic transform callback inside Field.apply — type-validation and __pg_accept__ coercions are unaffected. None resolves to enable_init (init → True, non-init → False) at this layer, so non-init fields skip wrapping even when enable_validation is manually re-enabled. (pg.Object's sym=False materializes an explicit False onto init fields, overriding this default.)

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 
961
962
963
964
965
966
967
968
969
970
971
972
973
@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 
980
981
982
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 
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
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 
1042
1043
1044
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 
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
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 
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
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 
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
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 ``sym=True`` subclass of a ``sym=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
    if self._enable_wrap is None and base_field._enable_wrap is False:
        self._enable_wrap = 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

Apply current field to a value, which validate and complete the value.

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

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 
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
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:
    """Apply current field to a value, which validate and complete the value.

    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.

    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 
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
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),
            ('enable_wrap', self._enable_wrap, 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 
151
152
153
154
155
156
157
158
159
160
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 
162
163
164
165
166
167
168
169
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 
210
211
212
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 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
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 an 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(value_spec=pg.Dict([
   ('x', pg.typing.Float(min_value=0.0)),
   (pg.typing.StrKey('foo.*'), pg.typing.Int())
]))

You may noticed that the code above pass a string 'x' for 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 either use 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 
106
107
108
@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

An KeySpec object derived from this key spec by extending the base.
Source code in pygx/typing/_class_schema.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@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:
      An ``KeySpec`` object derived from this key spec by extending the base.
    """

from_str classmethod 

from_str(key: str) -> Self

Get a concrete ValueSpec from annotation.

Source code in pygx/typing/_class_schema.py 
127
128
129
130
131
@classmethod
def from_str(cls, key: str) -> Self:  # pragma: no cover
    """Get a concrete ValueSpec from annotation."""
    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: 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: pg.typing.Int(max_value=5, default=2) z: pg.typing.Str('foo').freeze()

assert 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. If required a field is missing from the schema, and the object's allow_partial is set to False, a KeyError will raise. Otherwise a partially validated/transformed dict will be returned. Missing values in the object will be placeheld by :const:pygx.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 
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
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 appeared onces will be used in the merged schema.

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

(Optinoal) description of the schema.

 None

Returns:

Type Description
Schema

The merged schema.
Source code in pygx/typing/_class_schema.py 
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
@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 appeared onces will be
    used in the merged schema.

    Args:
      schema_list: A list of schemas to merge.
      name: (Optional) name of the merged schema.
      description: (Optinoal) 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 
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
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 that is acceptable to the other schema is acceptable to current schema.

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

Source code in pygx/typing/_class_schema.py 
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
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 that is acceptable to the other schema is acceptable to
        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

string as input key.

 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 
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
def get_field(self, key: str | int) -> Field | None:
    """Get field definition (Field) for a key.

    Args:
      key: string as input key.

    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 
1898
1899
1900
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 
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
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_enable_wrap: bool = False,
    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_enable_wrap bool

When True, child_transform is applied only to fields whose enable_wrap is True (and not threaded into the subtree of an enable_wrap=False field) — matching the pg.Object construct's external per-field gating, so an enable_wrap=False container lands verbatim. Defaults to False (the transform applies uniformly to every field). Only consulted on the resolve path, which is the only path a child_transform takes.

 False
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 
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
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_enable_wrap: bool = False,
    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::

          ------------------------------------------------------------
                                  |   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_enable_wrap: When True, ``child_transform`` is applied only to
        fields whose ``enable_wrap`` is True (and not threaded into the
        subtree of an ``enable_wrap=False`` field) — matching the
        ``pg.Object`` construct's external per-field gating, so an
        ``enable_wrap=False`` container lands verbatim. Defaults to False
        (the transform applies uniformly to every field). 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_enable_wrap,
            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 dict object is conformed with the schema.

Source code in pygx/typing/_class_schema.py 
2353
2354
2355
2356
2357
2358
2359
2360
2361
2362
2363
2364
def validate(
    self,
    dict_obj: dict[str, Any],
    allow_partial: bool = False,
    root_path: topology.KeyPath | None = None,
) -> None:
    """Validates whether dict object is conformed with 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 
2371
2372
2373
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 
2393
2394
2395
2396
2397
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]

Return an iteratable of KeySpecs in declaration order.

Source code in pygx/typing/_class_schema.py 
2399
2400
2401
def keys(self) -> Iterable[KeySpec]:
    """Return an iteratable 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 
2403
2404
2405
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 
2407
2408
2409
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 
2416
2417
2418
2419
2420
2421
2422
2423
2424
2425
2426
2427
2428
2429
2430
2431
2432
2433
2434
2435
2436
2437
2438
2439
2440
2441
2442
2443
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 
2458
2459
2460
2461
2462
2463
2464
2465
2466
2467
2468
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 
2478
2479
2480
2481
2482
2483
2484
2485
2486
2487
@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 | .apply | | validation and | | | transformation | | +-----------------------+-------------------------------------------------+ | 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.

ValueSpec object can also be derived from annotations. For example, the class below uses explicit value specs::

class A(pg.Object): a: pg.typing.List(pg.typing.Str()) b: pg.typing.Dict().set_default({'key': 'value'}) c: pg.typing.List(pg.typing.Any()).noneable() x: pg.typing.Int() y: pg.typing.Str().noneable() z: pg.typing.Union([pg.typing.Int(), pg.typing.Float()])

which can equivalently be written using 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

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 NO_FAST_APPLY (or any FastApplyInfo() with is_eligible=False).

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 referenes 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 
541
542
543
544
545
@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 
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
@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 
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
@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 post-validation user transform callable and returns self.

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

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 deep copy of the static default. 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 
626
627
628
629
630
631
632
633
634
635
636
637
638
@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 deep copy of the static default. 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 symoblic attribute and want 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 
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
@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
    symoblic attribute and want 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 
689
690
691
692
693
694
695
696
697
698
699
@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 
701
702
703
704
705
706
707
708
709
710
711
712
713
@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). Choose the default value if the input value is MISSING_VALUE (2). Check whether the input value is None. (2.a) Input value is None and value_spec.is_noneable() is False, raises Error. (2.b) Input value is not None or value_spec.is_noneable() is True, goto step (3). (3). Run value_spec.custom_apply if the input value is a CustomTyping instance. (3.a). If value_spec.custom_apply returns a value that indicates to proceed with standard apply, goto step (4). (3.b). Else goto step (6) (4). Check the input value type against the value_spec.value_type. (4.a). If their value type matches, go to step (5) (4.b). Else if the value spec's value type accepts the input via a __pg_accept__(value) classmethod, perform the coercion and go to step (5). (4.c) Otherwise raises type mismatch. (5). Perform type-specific and user validation and transformation. For complex types such as Dict, List, Tuple, call child_spec.apply recursively on the child fields. (6). Perform user transform and returns final value (invoked at Field.apply.)

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

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 
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
@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``::

       (1). Choose the default value if the input value is ``MISSING_VALUE``
       (2). Check whether the input value is None.
         (2.a) Input value is None and ``value_spec.is_noneable()`` is False,
               raises Error.
         (2.b) Input value is not None or ``value_spec.is_noneable()`` is True,
               goto step (3).
       (3). Run ``value_spec.custom_apply`` if the input value is a
            ``CustomTyping`` instance.
         (3.a). If ``value_spec.custom_apply`` returns a value that indicates to
                proceed with standard apply, goto step (4).
         (3.b). Else goto step (6)
       (4). Check the input value type against the ``value_spec.value_type``.
         (4.a). If their value type matches, go to step (5)
         (4.b). Else if the value spec's value type accepts the input via a
                ``__pg_accept__(value)`` classmethod, perform the coercion
                and go to step (5).
         (4.c)  Otherwise raises type mismatch.
       (5). Perform type-specific and user validation and transformation.
            For complex types such as Dict, List, Tuple, call `child_spec.apply`
            recursively on the child fields.
       (6). Perform user transform and returns final value
            (invoked at Field.apply.)

    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.

    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 
789
790
791
792
793
794
795
796
797
798
799
@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 
801
802
803
804
805
806
807
808
809
810
811
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 
813
814
815
816
817
818
819
820
821
822
@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 of custom value type.

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

As a result, CustomTyping makes the schema system extensible without modifying existing value specs. For example, value generators can extend CustomTyping and be assignable to any fields.

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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@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,
)

Docstring.

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 
100
101
102
103
104
105
106
107
108
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 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@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 
70
71
72
73
74
75
76
77
78
79
80
81
82
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 
94
95
96
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 
 98
 99
100
101
102
103
104
105
106
107
108
@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.

Example::

# Create a key spec that specifies list items from 1 to 5 (zero-based).
key = pg.typing.ListKey(min_value=1, max_value=5)

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

Constructor.

Parameters:

Name Type Description Default
min_value int

Min value that is acceptable for the list index.

 0
max_value int | None

Max value that is acceptable for the list index. If None, there is no upper bound for list index.

 None
Source code in pygx/typing/_key_specs.py 
238
239
240
241
242
243
244
245
246
247
248
def __init__(self, min_value: int = 0, max_value: int | None = None):
    """Constructor.

    Args:
      min_value: Min value that is acceptable for the list index.
      max_value: Max value that is acceptable for the list index. If None, there
        is no upper bound for list index.
    """
    super().__init__()
    self._min_value = min_value
    self._max_value = max_value

min_value property 

min_value: int

Returns min value of acceptable list index value.

max_value property 

max_value: int | None

Returns max value of acceptable list index value.

extend 

extend(base: KeySpec) -> Self

Extend current key spec on top of base spec.

Source code in pygx/typing/_key_specs.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
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 
279
280
281
282
283
284
285
def match(self, key: Any) -> bool:
    """Returns whether this key spec can match against input key."""
    return (
        isinstance(key, int)
        and (self._min_value <= key)
        and (not self._max_value 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 
287
288
289
290
291
292
293
294
295
296
297
298
299
@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 key.

Subclasses: pygx.typing.StrKey.

StrKey 

StrKey(regex: str | None = None)

Bases: NonConstKey

Class that represents a variable string key.

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 
163
164
165
166
167
168
169
170
171
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 
173
174
175
176
177
178
179
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 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@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 
336
337
338
339
340
341
342
343
344
345
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 
347
348
349
350
351
352
353
354
355
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 
362
363
364
365
366
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 
368
369
370
371
372
373
374
375
376
377
378
@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 'TupleKey(index={self._index})'

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.

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 required 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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
4499
4500
4501
4502
4503
4504
4505
4506
4507
4508
4509
4510
4511
4512
4513
4514
4515
4516
4517
4518
4519
4520
4521
4522
4523
4524
4525
4526
4527
4528
4529
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 is not accceptable.
    """
    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 
4535
4536
4537
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 
4539
4540
4541
4542
4543
4544
4545
4546
4547
4548
4549
4550
4551
4552
4553
4554
4555
4556
4557
4558
4559
4560
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 
4562
4563
4564
4565
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.

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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
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 is not accceptable.
    """
    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.

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()

# An 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.

Source code in pygx/typing/_value_specs.py 
3398
3399
3400
3401
3402
3403
3404
3405
3406
3407
3408
3409
3410
3411
3412
3413
3414
3415
3416
3417
3418
3419
3420
3421
3422
3423
3424
3425
3426
3427
3428
3429
3430
3431
3432
3433
3434
3435
3436
3437
3438
3439
3440
3441
3442
3443
3444
3445
3446
3447
3448
3449
3450
3451
3452
3453
3454
3455
3456
3457
3458
3459
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 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 
3465
3466
3467
3468
3469
3470
3471
3472
3473
3474
3475
3476
@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 
3675
3676
3677
3678
3679
3680
3681
3682
3683
3684
3685
3686
3687
3688
3689
3690
3691
3692
3693
3694
3695
3696
3697
3698
3699
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]
        | 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,
)

Bases: Generic, ValueSpecBase

Value spec for dict type.

Examples::

# A required symbolic 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] | 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 value spec. If None, it specifies a schema-less Dict that may accept arbitrary key/value pairs. 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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
2811
2812
2813
2814
2815
2816
2817
2818
2819
2820
2821
2822
2823
2824
2825
2826
2827
2828
2829
2830
2831
2832
2833
2834
2835
2836
2837
2838
2839
2840
2841
2842
2843
2844
2845
2846
2847
2848
2849
2850
2851
2852
2853
2854
2855
2856
2857
2858
2859
2860
2861
2862
2863
2864
2865
2866
2867
2868
2869
2870
2871
2872
2873
2874
2875
2876
2877
2878
2879
2880
2881
2882
2883
2884
2885
2886
2887
2888
2889
2890
2891
2892
2893
2894
2895
2896
2897
def __init__(
    self,
    schema: (
        class_schema.ValueSpec
        | Schema
        | dict[class_schema.FieldKeyDef, class_schema.FieldValueDef]
        | list[Field | class_schema.FieldDef]
        | 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,
):  # 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 value
        spec.
        If None, it specifies a schema-less Dict that may accept arbitrary
        key/value pairs.
        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 is not accceptable.
    """
    if schema is not None:
        if 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)
    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
            )

    # Automatically generate default based on schema if default is not present.
    if MISSING_VALUE == default:
        self.set_default(default)

schema property 

schema: Schema | None

Returns the schema of this dict spec.

noneable 

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

Override noneable in Dict to always set default value None.

Source code in pygx/typing/_value_specs.py 
2957
2958
2959
2960
2961
2962
2963
2964
2965
2966
2967
2968
def noneable(
    self, is_noneable: bool = True, use_none_as_default: bool = True
) -> Self:
    """Override noneable in Dict to always set default value None."""
    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 
2984
2985
2986
2987
2988
2989
2990
2991
2992
2993
@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 
3093
3094
3095
3096
3097
3098
3099
3100
3101
3102
3103
3104
3105
3106
3107
3108
3109
3110
3111
3112
3113
3114
3115
3116
3117
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,
            ),
            ('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.

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')

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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
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 is not accceptable.
    """
    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

Noneable is specially treated for Enum.

Source code in pygx/typing/_value_specs.py 
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
def noneable(
    self, is_noneable: bool = True, use_none_as_default: bool = True
) -> Self:
    """Noneable is specially treated for Enum."""
    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 
1678
1679
1680
1681
1682
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 
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
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.

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 is not accceptable.

 False
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 
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
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 is not accceptable.
      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.

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.

Source code in pygx/typing/_value_specs.py 
3764
3765
3766
3767
3768
3769
3770
3771
3772
3773
3774
3775
3776
3777
3778
3779
3780
3781
3782
3783
3784
3785
3786
3787
3788
3789
3790
3791
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."""
    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

Generic value spec.

GenericMeta

Bases: ABCMeta

Metaclass for generic value spec.

GenericTypeAlias 

GenericTypeAlias(*args, **kwargs)

Bases: Generic

Base class for generic type aliases.

Source code in pygx/typing/_value_specs.py 
4437
4438
4439
4440
4441
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::

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.

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 is not accceptable.

 False
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 
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
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 is not accceptable.
      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
)

Bases: Generic, ValueSpecBase

Value spec for list type.

Examples::

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

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

# An 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 is not accceptable.

 False
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 
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
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,
):  # 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 is not accceptable.
      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(
        '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',
    )
    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.

forward_refs property 

forward_refs: set[ForwardRef]

Returns forward references used in this spec.

min_size property 

min_size: int

Returns max 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 
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
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),
            ('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 value of acceptable values.

 None
max_value int | 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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
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 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 is not accceptable.
    """
    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 
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
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.

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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
3174
3175
3176
3177
3178
3179
3180
3181
3182
3183
3184
3185
3186
3187
3188
3189
3190
3191
3192
3193
3194
3195
3196
3197
3198
3199
3200
3201
3202
3203
3204
3205
3206
3207
3208
3209
3210
3211
3212
3213
3214
3215
3216
3217
3218
3219
3220
3221
3222
3223
3224
3225
3226
3227
3228
3229
3230
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 is not accceptable.
    """
    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 
3276
3277
3278
3279
3280
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 
3317
3318
3319
3320
3321
3322
3323
3324
3325
3326
3327
3328
3329
3330
3331
3332
3333
3334
3335
3336
3337
3338
3339
3340
3341
3342
3343
3344
3345
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:
        value_type = self._value_type
        assert isinstance(value_type, type)
        name = value_type.__name__
    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

Optional.

Source code in pygx/typing/_value_specs.py 
4437
4438
4439
4440
4441
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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
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 is not accceptable.
    """
    super().__init__(
        value_type,
        default,
        default_factory=default_factory,
        is_noneable=is_noneable,
        frozen=frozen,
        strict=strict,
    )

Sequence 

Sequence(*args, **kwargs)

Bases: GenericTypeAlias

Sequence.

Source code in pygx/typing/_value_specs.py 
4437
4438
4439
4440
4441
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.

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.

 None
is_noneable bool

If True, None is acceptable.

 False
frozen bool

If True, values other than the default value is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
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.
      is_noneable: If True, None is acceptable.
      frozen: If True, values other than the default value is not accceptable.
    """
    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 
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
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.

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)])

# A 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 list. Applicable only for variable-length tuple.

 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], 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 is not accceptable.

 False
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 
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315
2316
2317
2318
2319
2320
2321
2322
2323
2324
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343
2344
2345
2346
2347
2348
2349
2350
2351
2352
2353
2354
2355
2356
2357
2358
2359
2360
2361
2362
2363
2364
2365
2366
2367
2368
2369
2370
2371
2372
2373
2374
2375
2376
2377
2378
2379
2380
2381
2382
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 list. Applicable only for variable-length
        tuple.
      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 is not accceptable.
      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 max 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 
2453
2454
2455
2456
2457
2458
2459
2460
@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 
2704
2705
2706
2707
2708
2709
2710
2711
2712
2713
2714
2715
2716
2717
2718
2719
2720
2721
2722
2723
2724
2725
2726
2727
2728
2729
2730
2731
2732
2733
2734
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.

Examples::

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

# An 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)

Source code in pygx/typing/_value_specs.py 
3824
3825
3826
3827
3828
3829
3830
3831
3832
3833
3834
3835
3836
3837
3838
3839
3840
3841
3842
3843
3844
3845
3846
3847
3848
3849
3850
3851
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
    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 
3865
3866
3867
3868
3869
3870
3871
@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 
3915
3916
3917
3918
3919
3920
3921
3922
3923
3924
3925
3926
3927
3928
3929
3930
3931
3932
3933
3934
3935
3936
3937
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.

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})

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 is not accceptable.

 False
Source code in pygx/typing/_value_specs.py 
3992
3993
3994
3995
3996
3997
3998
3999
4000
4001
4002
4003
4004
4005
4006
4007
4008
4009
4010
4011
4012
4013
4014
4015
4016
4017
4018
4019
4020
4021
4022
4023
4024
4025
4026
4027
4028
4029
4030
4031
4032
4033
4034
4035
4036
4037
4038
4039
4040
4041
4042
4043
4044
4045
4046
4047
4048
4049
4050
4051
4052
4053
4054
4055
4056
4057
4058
4059
4060
4061
4062
4063
4064
4065
4066
4067
4068
4069
4070
4071
4072
4073
4074
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 is not accceptable.
    """
    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_types = set()
    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):
            candidate_types.update(child_value_type)
        else:
            candidate_types.add(child_value_type)

    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 
4134
4135
4136
4137
4138
4139
4140
4141
@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

Customized noneable for Union.

Source code in pygx/typing/_value_specs.py 
4195
4196
4197
4198
4199
4200
4201
4202
4203
4204
def noneable(
    self, is_noneable: bool = True, use_none_as_default: bool = True
) -> Self:
    """Customized noneable for Union."""
    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 
4211
4212
4213
4214
4215
4216
4217
4218
4219
4220
4221
4222
4223
4224
4225
4226
4227
4228
4229
4230
4231
4232
4233
4234
4235
4236
4237
4238
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 
4343
4344
4345
4346
4347
4348
4349
4350
4351
4352
4353
4354
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 
4363
4364
4365
4366
4367
4368
4369
4370
4371
4372
4373
4374
4375
4376
4377
4378
4379
4380
4381
4382
4383
4384
4385
4386
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]
)

Helper class for invoking callable objects with optional keyword args.

Examples::

f = pg.typing.CallableWithOptionalKeywordArgs(lambda x: x ** 2, 'y') # Returns 4. Keyword 'y' is ignored. f(2, y=3)

Source code in pygx/typing/_callable_ext.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
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

Value placeholder for arguments whose value will be provided by presets.

Example:

def foo(x, y=pg.PresetArgValue(default=1)) return x + y

with pg.preset_args(y=2): print(foo(x=1)) # 3: y=2 print(foo(x=1)) # 2: y=1

Source code in pygx/typing/_callable_ext.py 
50
51
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 
75
76
77
78
79
80
81
82
83
84
@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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@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 
64
65
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 annotations from 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.

 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 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def annotation_from_str(
    annotation_str: str,
    parent_module: types.ModuleType | None = None,
) -> typing.Any:
    """Parses annotations from 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'
    ```

    Args:
      annotation_str: String form of type annotations. E.g. "list[str]"
      parent_module: The module where the annotation was defined.

    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 
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
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::

        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 description form docstrs.

 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 with key 'return_value'.

 False

Returns:

Type Description
Schema

A pg.typing.Schema object.
Source code in pygx/typing/_callable_signature.py 
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
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 description form docstrs.
      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 with
        key 'return_value'.

    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 signature from a python callable.

Source code in pygx/typing/_callable_signature.py 
917
918
919
920
921
922
923
def signature(
    func: Callable[..., Any],
    auto_typing: bool = True,
    auto_doc: bool = True,
) -> Signature:  # pylint:disable=g-bare-generic
    """Gets signature from a python callable."""
    return Signature.from_callable(func, auto_typing, auto_doc)

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_value 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 
2550
2551
2552
2553
2554
2555
2556
2557
2558
2559
2560
2561
2562
2563
2564
2565
2566
2567
2568
2569
2570
2571
2572
2573
2574
2575
2576
2577
2578
2579
2580
2581
2582
2583
2584
2585
2586
2587
2588
2589
2590
2591
2592
2593
2594
2595
2596
2597
2598
2599
2600
2601
2602
2603
2604
2605
2606
2607
2608
2609
2610
2611
2612
2613
2614
2615
2616
2617
2618
2619
2620
2621
2622
2623
2624
2625
2626
2627
2628
2629
2630
2631
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_value`` 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}'
        )
    return Field(key, value, description, field_metadata)

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_value 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 
2634
2635
2636
2637
2638
2639
2640
2641
2642
2643
2644
2645
2646
2647
2648
2649
2650
2651
2652
2653
2654
2655
2656
2657
2658
2659
2660
2661
2662
2663
2664
2665
2666
2667
2668
2669
2670
2671
2672
2673
2674
2675
2676
2677
2678
2679
2680
2681
2682
2683
2684
2685
2686
2687
2688
2689
2690
2691
2692
2693
2694
2695
2696
2697
2698
2699
2700
2701
2702
2703
2704
2705
2706
2707
2708
2709
2710
2711
2712
2713
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::

        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_value`` 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 structure docstring of a Python symbol.

Source code in pygx/typing/_docstr.py 
150
151
152
153
def docstr(symbol: Any) -> DocStr | None:
    """Gets structure docstring of a Python symbol."""
    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.

x and y are considered equal when they are the same

instance or have the same code (e.g. lambda x: x).

x and y are considered equal when:

static method: The same method from the same class hierarchy. E.g. subclass inherits a base class' static method. class method: The same method from the same class. Inherited class method are considered different class method. instance method: When self is not bound, the same method from the same class hierarchy (like static method). 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 optinoal function or method object.

 required

Returns:

Type Description
bool

Returns True if x and y are considered equal. Meaning that they are either the same instance or derived from the same code and have the same effect.
Source code in pygx/typing/_inspection.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
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. lambda x: x).

    For methods: `x` and `y` are considered equal when:
      static method: The same method from the same class hierarchy. E.g. subclass
        inherits a base class' static method.
      class method: The same method from the same class. Inherited class method
        are considered different class method.
      instance method: When `self` is not bound, the same method from the same
        class hierarchy (like static method). When `self` is bound, the same
        method on the same object.

    Args:
      x: An optional function or method object.
      y: An optinoal function or method object.

    Returns:
      Returns True if `x` and `y` are considered equal. Meaning that they are
        either the same instance or derived from the same code and have 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 outer class.

Example::

class A: pass

class A1: class B: class C: ...

pg.typing.outer_class(B) is A1 pg.typing.outer_class(C) is B pg.typing.outer_class(C, base_cls=A) is None pg.typing.outer_class(C, base_cls=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 of the outer class. If provided, an outer class that is not a subclass of base_cls will be returned as None.

 None
immediate bool

Whether to return the immediate outer class or a class in the nesting hierarchy that is 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 cannot find one or the outer class is not a subclass of base_cls.
Source code in pygx/typing/_inspection.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def get_outer_class(
    cls: type[Any],
    base_cls: type[Any] | tuple[type[Any], ...] | None = None,
    immediate: bool = False,
) -> type[Any] | None:
    """Returns the outer class.

    Example::

      class A:
        pass

      class A1:
        class B:
          class C:
            ...

      pg.typing.outer_class(B) is A1
      pg.typing.outer_class(C) is B
      pg.typing.outer_class(C, base_cls=A) is None
      pg.typing.outer_class(C, base_cls=A1) is None

    Args:
      cls: The class to get the outer class for.
      base_cls: The base class of the outer class. If provided, an outer class
        that is not a subclass of `base_cls` will be returned as None.
      immediate: Whether to return the immediate outer class or a class in the
        nesting hierarchy that is a subclass of `base_cls`. Applicable when
        `base_cls` is not None.

    Returns:
      The outer class of `cls`. None if cannot find one or the outer class is
        not a subclass of `base_cls`.
    """
    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 
112
113
114
115
116
117
118
119
120
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.

Source code in pygx/typing/_inspection.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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."""
    if base is None:
        return typing.get_args(maybe_generic)
    else:
        orig_cls = typing.get_origin(maybe_generic)
        if orig_cls is not None:
            orig_bases = (maybe_generic,)
        else:
            orig_bases = getattr(maybe_generic, '__orig_bases__', ())
        for orig_base in orig_bases:
            if get_type(orig_base) is base:
                return typing.get_args(orig_base)
        return ()

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 
107
108
109
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 
102
103
104
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.

Source code in pygx/typing/_inspection.py 
26
27
28
def is_instance(value: Any, target: Any) -> bool:
    """An isinstance extension that supports Any and generic types."""
    return is_subclass(type(value), target)

is_subclass 

is_subclass(src: Any, target: Any) -> bool

An issubclass extension that supports Any and generic types.

Source code in pygx/typing/_inspection.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def is_subclass(src: Any, target: Any) -> bool:
    """An issubclass extension that supports Any and generic types."""

    def _is_subclass(src: type[Any], target: type[Any]) -> bool:
        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:
                # Both soure and target is not 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):
                if not _is_subclass(s_arg, t_arg):
                    return False
            return True
        else:
            # A class could inherit multiple generic types. However it does not
            # provide the type arguments for the target generic base. E.g.
            #
            # class A(Generic[X, Y]):
            # class B(A, Generic[X, Y]) :
            # B[int, int] is not a subclass of A[int, int].
            return False

    if isinstance(target, tuple):
        return any(_is_subclass(src, x) for x in target)
    return _is_subclass(src, target)

ensure_value_spec 

ensure_value_spec(
    value_spec: ValueSpec, src_spec: ValueSpec, root_path: KeyPath | None = None
) -> ValueSpec | None

Extract counter part from value spec that matches dest spec type.

Parameters:

Name Type Description Default
value_spec ValueSpec

Value spec.

 required
src_spec ValueSpec

Destination value spec.

 required
root_path KeyPath | None

An optional path for the value to include in error message.

 None

Returns:

Type Description
ValueSpec | None

value_spec of src_spec_type

Raises:

Type Description
TypeError

When value_spec cannot match src_spec_type.
Source code in pygx/typing/_value_specs.py 
4763
4764
4765
4766
4767
4768
4769
4770
4771
4772
4773
4774
4775
4776
4777
4778
4779
4780
4781
4782
4783
4784
4785
4786
4787
4788
4789
4790
4791
4792
4793
def ensure_value_spec(
    value_spec: class_schema.ValueSpec,
    src_spec: class_schema.ValueSpec,
    root_path: topology.KeyPath | None = None,
) -> class_schema.ValueSpec | None:
    """Extract counter part from value spec that matches dest spec type.

    Args:
      value_spec: Value spec.
      src_spec: Destination value spec.
      root_path: An optional path for the value to include in error message.

    Returns:
      value_spec of src_spec_type

    Raises:
      TypeError: When value_spec cannot match src_spec_type.
    """
    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 for functions that maybe use preset argument values.

Usage::

@pg.typing.enable_preset_args def foo(x, y=pg.typing.PresetArgValue(default=1)): return x + y

with pg.typing.preset_args(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 include all preset kwargs (even not makred as PresetArgValue) when callng the function.

 False
preset_name str

The name of the preset to specify kwargs.

 'global'

Returns:

Type Description
Callable[[Callable[..., Any]], Callable[..., Any]]

A decorated function that could consume the preset argument values.
Source code in pygx/typing/_callable_ext.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def enable_preset_args(
    include_all_preset_kwargs: bool = False, preset_name: str = 'global'
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator for functions that maybe use preset argument values.

    Usage::

      @pg.typing.enable_preset_args
      def foo(x, y=pg.typing.PresetArgValue(default=1)):
        return x + y

      with pg.typing.preset_args(y=2):
        print(foo(x=1))  # 3: y=2
      print(foo(x=1))  # 2: y=1

    Args:
      include_all_preset_kwargs: Whether to include all preset kwargs (even
        not makred as `PresetArgValue`) when callng the function.
      preset_name: The name of the preset to specify kwargs.

    Returns:
      A decorated function that could consume the preset argument values.
    """

    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[dict[str, Any]]

Context manager to enable calling with user kwargs.

Parameters:

Name Type Description Default
kwargs dict[str, Any]

The preset kwargs to be used by preset-enabled functions within the context.

 required
preset_name str

The name of the preset to specify kwargs. enable_preset_args allows users to pass a preset name, which will be used to identify the present to be used.

 '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
dict[str, Any]

Current preset kwargs.
Source code in pygx/typing/_callable_ext.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@contextlib.contextmanager
def preset_args(
    kwargs: dict[str, Any],
    *,
    preset_name: str = 'global',
    inherit_preset: str | bool = False,
) -> Iterator[dict[str, Any]]:
    """Context manager to enable calling with user kwargs.

    Args:
      kwargs: The preset kwargs to be used by preset-enabled functions within
        the context.
      preset_name: The name of the preset to specify kwargs.
        `enable_preset_args` allows users to pass a preset name, which will be
        used to identify the present to be used.
      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:
      Current preset kwargs.
    """

    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 value spec to JSON schema.

Source code in pygx/typing/_json_schema.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
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 value spec to JSON schema."""
    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