pygx.wire

Wire formats for serializing PyGX values.

wire

Wire formats for serializing PyGX values.

pygx.wire turns PyGX values into transport- and storage-friendly representations and back. It is built in three decoupled layers:

• The canonical representation: a tree of plain Python values (WireValue, == the JSON data model). Both halves below meet only through this plain-data contract.
• The object protocol: WireConvertible — an object that converts itself to/from a WireValue via to_json / from_json. WireConversionContext carries shared state (e.g. ref tracking) across a single conversion.
• The codecs: WireFormat implementations that encode/decode a WireValue to bytes/text. JsonFormat is the default; YamlFormat is a second format with different capabilities. Pick one via serialize / deserialize or format= on the symbolic save/load helpers; add your own with register_format.

WireConversionContext

WireConversionContext(supports_refs: bool = True)

Bases: WireConvertible

JSON conversion context.

WireConversionContext is introduced to handle serialization scenarios where operations cannot be performed in a single pass. For example: Serialization and deserialization of shared objects across different locations.

Shared object serialization/deserialization.

In PyGX, only values referenced by pg.Ref and non-PyGX managed objects are sharable. This ensures that multiple references to the same object are serialized only once. During deserialization, the object is created just once and shared among all references.

Source code in pygx/wire/_conversion.py

def __init__(
    self,
    supports_refs: bool = True,
) -> None:
    self._shared_objects: list[WireConversionContext.ObjectEntry] = []
    self._id_to_shared_object = {}
    # Tracks whether any entry needs the post-pass (`_maybe_deref`) tree
    # walk to fix up index rewrites or unwrap non-dict singletons. When
    # False after serialization, `to_json` can return the root as-is.
    self._needs_post_pass = False
    # Whether the target wire format can represent shared/cyclic references
    # (via `__ref__` + a shared-object table). When False, a value that is
    # referenced more than once raises rather than emitting a ref.
    self._supports_refs = supports_refs

ObjectEntry dataclass

ObjectEntry(
    value: Any,
    serialized: WireValue | None,
    ref_index: int,
    ref_count: int,
    inlined_at: dict | None = None,
)

Per-value bookkeeping for serialize_maybe_shared.

Allocated once per non-primitive value on the hot to_json path, so slots=True matters: it skips the __dict__ and shrinks the per-entry footprint.

inlined_at is non-None when the entry's serialized form is currently inlined into the json tree at that dict (same object). On a second reference, we mutate that dict in place to a {REF_KEY: idx} wrap and move the original content to serialized. inlined_at is None when the entry has already been promoted to a ref, or was wrapped from the start (e.g. list-shaped serialized that we can't mutate to a dict in place).

get_shared

get_shared(ref_index: int) -> ObjectEntry

Gets the shared object of a ref index.

Source code in pygx/wire/_conversion.py

def get_shared(self, ref_index: int) -> ObjectEntry:
    """Gets the shared object of a ref index."""
    return self._shared_objects[ref_index]

next_shared_index

next_shared_index() -> int

Returns the next shared index.

Source code in pygx/wire/_conversion.py

def next_shared_index(self) -> int:
    """Returns the next shared index."""
    return len(self._shared_objects)

serialize_maybe_shared

serialize_maybe_shared(
    value: Any, json_fn: Callable[..., WireValue] | None = None, **kwargs
) -> WireValue

Track maybe-shared objects and return their JSON representation.

Inline-first strategy

• First encounter, dict-shaped: return the serialized content directly and remember it as the "inline" location.
• First encounter, other shape (list, scalar): wrap in {REF_KEY: idx} — we can't mutate non-dicts to a ref shape later, so post-pass unwrap handles ref_count == 1.
• Subsequent encounter: mutate the (still-inline) first occurrence in place into {REF_KEY: idx}, stash the original content on the entry, and return a fresh {REF_KEY: idx}.

This avoids the wrap/unwrap dance in the common (no-share) case and preserves identity-preserving dedup when sharing actually occurs.

Callers MUST nest the returned value (e.g. parent[k] = result); a later .update(result) would mutate an orphan dict on promotion and the ref would never appear in the tree.

Note: context is always a named arg on the callers (utils.to_json, Ref.sym_jsonify), never inside **kwargs, so we don't bother filtering it out of kwargs here.

Source code in pygx/wire/_conversion.py

def serialize_maybe_shared(
    self,
    value: Any,
    json_fn: Callable[..., WireValue] | None = None,
    **kwargs,
) -> WireValue:
    """Track maybe-shared objects and return their JSON representation.

    Inline-first strategy:
      * **First encounter, dict-shaped:** return the serialized content
        directly and remember it as the "inline" location.
      * **First encounter, other shape (list, scalar):** wrap in
        ``{REF_KEY: idx}`` — we can't mutate non-dicts to a ref shape
        later, so post-pass unwrap handles ref_count == 1.
      * **Subsequent encounter:** mutate the (still-inline) first
        occurrence in place into ``{REF_KEY: idx}``, stash the original
        content on the entry, and return a fresh ``{REF_KEY: idx}``.

    This avoids the wrap/unwrap dance in the common (no-share) case and
    preserves identity-preserving dedup when sharing actually occurs.

    Callers MUST *nest* the returned value (e.g. ``parent[k] = result``);
    a later ``.update(result)`` would mutate an orphan dict on promotion
    and the ref would never appear in the tree.

    Note: `context` is always a named arg on the callers
    (`utils.to_json`, `Ref.sym_jsonify`), never inside `**kwargs`, so we
    don't bother filtering it out of `kwargs` here.
    """
    if json_fn is None:
        # pylint: disable-next=unnecessary-lambda-assignment
        json_fn = lambda **kwargs: to_json(value, **kwargs)
    value_id = id(value)

    # --- Subsequent encounter -------------------------------------------
    entry = self._id_to_shared_object.get(value_id)
    if entry is not None:
        if not self._supports_refs:
            raise ValueError(
                'This wire format cannot serialize shared/cyclic '
                f'references: {value!r} is referenced more than once. '
                'Use a format whose `supports_refs` is True (e.g. JSON), '
                'or remove the sharing (e.g. `pg.Ref`) from the value.'
            )
        inline = entry.inlined_at
        if inline is not None:
            # Promote: copy content out of the inline dict, then mutate it
            # in place to a ref pointing at the new shared entry.
            entry.serialized = dict(inline)
            inline.clear()
            inline[WireConvertible.REF_KEY] = entry.ref_index
            entry.inlined_at = None
        entry.ref_count += 1
        self._needs_post_pass = True
        return {WireConvertible.REF_KEY: entry.ref_index}

    # --- First encounter ------------------------------------------------
    serialized = json_fn(context=self, **kwargs)

    # `json_fn` may have re-entered `serialize_maybe_shared` on the same
    # value (e.g. the plain-dict path calls `to_json` which calls us
    # again). If the inner frame registered the entry, this outer frame
    # is a pass-through.
    if value_id in self._id_to_shared_object:
        return serialized

    # Register the entry. Dict-shaped serializations stay inline at the
    # caller's location (`inlined_at` points to the same dict). Other
    # shapes get a `{REF_KEY: idx}` wrap that the post-pass will unwrap.
    is_inline = isinstance(serialized, dict)
    entry = self.ObjectEntry(
        value=value,
        serialized=serialized,
        ref_index=len(self._shared_objects),
        ref_count=1,
        inlined_at=serialized if is_inline else None,
    )
    self._shared_objects.append(entry)
    self._id_to_shared_object[value_id] = entry
    if is_inline:
        return serialized
    self._needs_post_pass = True
    return {WireConvertible.REF_KEY: entry.ref_index}

to_json

to_json(*, root: Any, **kwargs) -> WireValue

Serializes a root node with the context to JSON value.

Source code in pygx/wire/_conversion.py

def to_json(self, *, root: Any, **kwargs) -> WireValue:
    """Serializes a root node with the context to JSON value."""
    # Hot path: no entry was ever promoted and no non-dict wraps exist,
    # so the tree is already in its final form.
    if not self._needs_post_pass:
        return root

    # Build the (pre-prune) index → (post-prune) index map. Entries with
    # ref_count == 1 are dropped from the emitted `shared_objects` array
    # because their content is (or will be) inlined at the call site.
    ref_index_map = {}
    kept_serialized = []
    for i, entry in enumerate(self._shared_objects):
        ref_index_map[i] = len(kept_serialized)
        if entry.ref_count != 1:
            kept_serialized.append(entry.serialized)

    root = self._maybe_deref(root, ref_index_map)
    if not kept_serialized:
        return root

    return {
        WireConvertible.CONTEXT_KEY: {
            'shared_objects': [
                self._maybe_deref(x, ref_index_map) for x in kept_serialized
            ],
        },
        WireConvertible.ROOT_VALUE_KEY: root,
    }

from_json classmethod

from_json(json_value: WireValue, **kwargs) -> WireConversionContext

Deserializes a WireConvertible value from JSON value.

Source code in pygx/wire/_conversion.py

@classmethod
def from_json(
    cls, json_value: WireValue, **kwargs
) -> 'WireConversionContext':
    """Deserializes a WireConvertible value from JSON value."""
    context = cls()
    if isinstance(json_value, dict):
        # Shared objects are serialized in a bottom-up order, thus dependent
        # shared objects must be deserialized first.
        if shared_objects_json := json_value.get('shared_objects'):
            for v in shared_objects_json:
                context.add_shared(
                    cls.ObjectEntry(
                        value=from_json(v, context=context, **kwargs),
                        serialized=v,
                        ref_index=context.next_shared_index(),
                        ref_count=0,
                    )
                )
    return context

WireConvertible

Interface for objects convertible to/from the canonical wire representation.

A wire-convertible object is one that can turn itself into a WireValue (a tree of plain Python dict/list/primitive values) and back, hence can be serialized by any WireFormat (JSON, YAML, ...). The representation coincides with the JSON data model, which is why the protocol methods are named to_json/from_json.

Subclasses of WireConvertible should implement:

• to_json: A method that returns a plain Python dict with a _type property whose value should identify the class.
• from_json: A class method that takes a plain Python dict and returns an instance of the class.

Example::

class MyObject(pg.WireConvertible):

def __init__(self, x: int):
  self.x = x

def to_json(self, **kwargs):
  return {
    '_type': 'MyObject',
    'x': self.x
  }

@classmethod
def from_json(cls, json_value, **kwargs):
  return cls(json_value['x'])

All symbolic types (see pygx.Symbolic) are JSON convertible.

from_json classmethod

from_json(json_value: WireValue, **kwargs: Any) -> Self

Creates an instance of this class from a plain Python value.

NOTE(daiyip): pg.Symbolic overrides from_json class method.

Parameters:

Name	Type	Description	Default
json_value	WireValue	JSON value type.	required
**kwargs	Any	Keyword arguments as flags to control object creation.	{}

Returns:

Type	Description
Self	An instance of cls.

Source code in pygx/wire/_conversion.py

@classmethod
def from_json(cls, json_value: WireValue, **kwargs: Any) -> Self:
    """Creates an instance of this class from a plain Python value.

    NOTE(daiyip): ``pg.Symbolic`` overrides ``from_json`` class method.

    Args:
      json_value: JSON value type.
      **kwargs: Keyword arguments as flags to control object creation.

    Returns:
      An instance of cls.
    """
    assert isinstance(json_value, dict)
    init_args = {
        k: from_json(v, **kwargs)
        for k, v in json_value.items()
        if k != WireConvertible.TYPE_NAME_KEY
    }
    return cls(**init_args)

to_json abstractmethod

to_json(
    *, context: Optional[WireConversionContext] = None, **kwargs: Any
) -> WireValue

Returns a plain Python value as a representation for this object.

A plain Python value are basic python types that can be serialized into JSON, e.g: bool, int, float, str, dict (with string keys), list, tuple where the container types should have plain Python values as their values.

Parameters:

Name	Type	Description	Default
context	Optional[WireConversionContext]	JSON conversion context.	None
**kwargs	Any	Keyword arguments as flags to control JSON conversion.	{}

Returns:

Type	Description
WireValue	A plain Python value.

Source code in pygx/wire/_conversion.py

@abc.abstractmethod
def to_json(
    self,
    *,
    context: Optional['WireConversionContext'] = None,
    **kwargs: Any,
) -> WireValue:
    """Returns a plain Python value as a representation for this object.

    A plain Python value are basic python types that can be serialized into
    JSON, e.g: ``bool``, ``int``, ``float``, ``str``, ``dict`` (with string
    keys), ``list``, ``tuple`` where the container types should have plain
    Python values as their values.

    Args:
      context: JSON conversion context.
      **kwargs: Keyword arguments as flags to control JSON conversion.

    Returns:
      A plain Python value.
    """

register classmethod

register(
    type_name: str,
    subclass: type[WireConvertible],
    override_existing: bool = False,
) -> None

Registers a class with a type name.

The type name will be used as the key for class lookup during deserialization. A class can be registered with multiple type names, but a type name should be uesd only for one class.

Parameters:

Name	Type	Description	Default
type_name	str	A global unique string identifier for subclass.	required
subclass	type[WireConvertible]	A subclass of WireConvertible.	required
override_existing	bool	If True, override the class if the type name is already present in the registry. Otherwise an error will be raised.	False

Source code in pygx/wire/_conversion.py

@classmethod
def register(
    cls,
    type_name: str,
    subclass: type['WireConvertible'],
    override_existing: bool = False,
) -> None:
    """Registers a class with a type name.

    The type name will be used as the key for class lookup during
    deserialization. A class can be registered with multiple type names, but
    a type name should be uesd only for one class.

    Args:
      type_name: A global unique string identifier for subclass.
      subclass: A subclass of WireConvertible.
      override_existing: If True, override the class if the type name is
        already present in the registry. Otherwise an error will be raised.
    """
    cls._TYPE_REGISTRY.register(type_name, subclass, override_existing)

add_module_alias classmethod

add_module_alias(module: str, alias: str | Sequence[str]) -> None

Adds a module alias so previous serialized objects could be loaded.

Source code in pygx/wire/_conversion.py

@classmethod
def add_module_alias(cls, module: str, alias: str | Sequence[str]) -> None:
    """Adds a module alias so previous serialized objects could be loaded."""
    cls._TYPE_REGISTRY.add_module_alias(module, alias)

set_trusted_pickle_types classmethod

set_trusted_pickle_types(
    allowed_types: Iterable[str | type[Any]] | None,
) -> None

Sets the global allow-list of types that may be unpickled.

Opaque (non-WireConvertible) objects are serialized via pickle (see _OpaqueObject). Because pickle.loads can execute arbitrary code, deserializing untrusted JSON is unsafe by default. Restrict it by allow-listing exactly the types your data is expected to contain::

pg.WireConvertible.set_trusted_pickle_types([MyData, np.ndarray]) obj = pg.from_json_str(untrusted_json) # refuses anything else

Parameters:

Name	Type	Description	Default
allowed_types	Iterable[str | type[Any]] | None	An iterable of types or fully-qualified module.qualname strings that may be reconstructed from pickle. Pass None to remove the restriction (the default, unrestricted behavior).	required

Note

The allow-list governs the standard-pickle reconstruction path, where the pickle stream references the concrete type by name. Objects serialized via the optional cloudpickle fallback (lambdas, closures, local classes) reference cloudpickle's code-building reconstructors instead, which cannot be meaningfully sandboxed by a type allow-list — loading those requires trusting the payload.

Source code in pygx/wire/_conversion.py

@classmethod
def set_trusted_pickle_types(
    cls, allowed_types: Iterable[str | type[Any]] | None
) -> None:
    """Sets the global allow-list of types that may be unpickled.

    Opaque (non-`WireConvertible`) objects are serialized via ``pickle``
    (see `_OpaqueObject`). Because ``pickle.loads`` can execute arbitrary
    code, deserializing untrusted JSON is unsafe by default. Restrict it
    by allow-listing exactly the types your data is expected to contain::

      pg.WireConvertible.set_trusted_pickle_types([MyData, np.ndarray])
      obj = pg.from_json_str(untrusted_json)  # refuses anything else

    Args:
      allowed_types: An iterable of types or fully-qualified
        ``module.qualname`` strings that may be reconstructed from pickle.
        Pass ``None`` to remove the restriction (the default, unrestricted
        behavior).

    Note:
      The allow-list governs the standard-pickle reconstruction path,
      where the pickle stream references the concrete type by name. Objects
      serialized via the optional ``cloudpickle`` fallback (lambdas,
      closures, local classes) reference cloudpickle's code-building
      reconstructors instead, which cannot be meaningfully sandboxed by a
      type allow-list — loading those requires trusting the payload.
    """
    cls._TRUSTED_PICKLE_TYPES = _normalize_trusted_pickle_types(
        allowed_types
    )

trusted_pickle_types classmethod

trusted_pickle_types() -> frozenset[str] | None

Returns the global pickle allow-list, or None if unrestricted.

Source code in pygx/wire/_conversion.py

@classmethod
def trusted_pickle_types(cls) -> frozenset[str] | None:
    """Returns the global pickle allow-list, or ``None`` if unrestricted."""
    return cls._TRUSTED_PICKLE_TYPES

is_registered classmethod

is_registered(type_name: str) -> bool

Returns True if a type name is registered. Otherwise False.

Source code in pygx/wire/_conversion.py

@classmethod
def is_registered(cls, type_name: str) -> bool:
    """Returns True if a type name is registered. Otherwise False."""
    return cls._TYPE_REGISTRY.is_registered(type_name)

class_from_typename classmethod

class_from_typename(type_name: str) -> type[WireConvertible] | None

Gets the class for a registered type name.

Parameters:

Name	Type	Description	Default
type_name	str	A string as the global unique type identifier for requested class.	required

Returns:

Type	Description
type[WireConvertible] | None	A type object if registered, otherwise None.

Source code in pygx/wire/_conversion.py

@classmethod
def class_from_typename(
    cls, type_name: str
) -> type['WireConvertible'] | None:
    """Gets the class for a registered type name.

    Args:
      type_name: A string as the global unique type identifier for requested
        class.

    Returns:
      A type object if registered, otherwise None.
    """
    return cls._TYPE_REGISTRY.class_from_typename(type_name)

registered_types classmethod

registered_types() -> Iterable[tuple[str, type[WireConvertible]]]

Returns an iterator of registered (serialization key, class) tuples.

Source code in pygx/wire/_conversion.py

@classmethod
def registered_types(cls) -> Iterable[tuple[str, type['WireConvertible']]]:
    """Returns an iterator of registered (serialization key, class) tuples."""
    return cls._TYPE_REGISTRY.iteritems()

load_types_for_deserialization classmethod

load_types_for_deserialization(
    *types_to_deserialize: type[Any],
) -> ContextManager[dict[str, type[Any]]]

Context manager for loading unregistered types for deserialization.

Example::

class A(pg.Object, to_json_key=False): x: int

class B(A): y: str with pg.JSONConvertile.load_types_for_deserialization(A, B): pg.from_json_str(A(1).to_json_str()) pg.from_json_str(B(1, 'hi').to_json_str())

Parameters:

Name	Type	Description	Default
*types_to_deserialize	type[Any]	A list of types to be loaded for deserialization.	()

Returns:

Type	Description
ContextManager[dict[str, type[Any]]]	A context manager within which the objects of the requested types could be deserialized.

Source code in pygx/wire/_conversion.py

@classmethod
def load_types_for_deserialization(
    cls, *types_to_deserialize: type[Any]
) -> ContextManager[dict[str, type[Any]]]:
    """Context manager for loading unregistered types for deserialization.

    Example::

      class A(pg.Object, to_json_key=False):
        x: int

      class B(A):
        y: str
      with pg.JSONConvertile.load_types_for_deserialization(A, B):
          pg.from_json_str(A(1).to_json_str())
          pg.from_json_str(B(1, 'hi').to_json_str())

    Args:
      *types_to_deserialize: A list of types to be loaded for deserialization.

    Returns:
      A context manager within which the objects of the requested types
        could be deserialized.
    """
    return cls._TYPE_REGISTRY.load_types_for_deserialization(
        *types_to_deserialize
    )

to_json_dict classmethod

to_json_dict(
    fields: dict[str, tuple[Any, Any] | Any],
    *,
    exclude_default=False,
    exclude_keys: set[str] | None = None,
    **kwargs
) -> dict[str, WireValue]

Helper method to create JSON dict from class and field.

Source code in pygx/wire/_conversion.py

@classmethod
def to_json_dict(
    cls,
    fields: dict[str, tuple[Any, Any] | Any],
    *,
    exclude_default=False,
    exclude_keys: set[str] | None = None,
    **kwargs,
) -> dict[str, WireValue]:
    """Helper method to create JSON dict from class and field."""
    json_dict: dict[str, WireValue] = {
        WireConvertible.TYPE_NAME_KEY: _serialization_key(cls)
    }
    exclude_keys = exclude_keys or set()
    if exclude_default:
        for k, (v, default) in fields.items():
            if k not in exclude_keys and v != default:
                json_dict[k] = to_json(v, **kwargs)
    else:
        json_dict.update(
            {
                k: to_json(v, **kwargs)
                for k, v in fields.items()
                if k not in exclude_keys
            }
        )
    return json_dict

JsonFormat

Bases: WireFormat

The built-in JSON wire format (the default).

WireFormat

Bases: ABC

Base class for a wire format: a codec over the canonical representation.

Subclasses set :attr:name and override :meth:encode / :meth:decode. They may flip the capability flags to declare which PyGX semantics they can represent; the conversion engine consults them (see supports_refs).

encode abstractmethod

encode(value: WireValue, *, indent: int | None = None) -> str | bytes

Encodes a WireValue tree into the format's serialized form.

Source code in pygx/wire/_format.py

@abc.abstractmethod
def encode(
    self, value: WireValue, *, indent: int | None = None
) -> str | bytes:
    """Encodes a ``WireValue`` tree into the format's serialized form."""

decode abstractmethod

decode(data: str | bytes) -> WireValue

Decodes the format's serialized form back into a WireValue tree.

Source code in pygx/wire/_format.py

@abc.abstractmethod
def decode(self, data: str | bytes) -> WireValue:
    """Decodes the format's serialized form back into a ``WireValue`` tree."""

YamlFormat

Bases: WireFormat

A YAML wire format -- a worked example of a second format.

YAML carries non-string keys natively (native_int_keys=True), so it skips JSON's int-key escaping. It does not represent shared/cyclic references (supports_refs=False): serializing a value with such sharing raises a clear error instead of emitting an anchor-less __ref__ table.

from_json

from_json(
    json_value: WireValue,
    *,
    context: WireConversionContext | None = None,
    auto_import: bool = True,
    convert_unknown: bool = False,
    **kwargs: Any
) -> Any

Deserializes a (maybe) WireConvertible value from JSON value.

Parameters:

Name	Type	Description	Default
json_value	WireValue	Input JSON value.	required
context	WireConversionContext | None	Serialization context.	None
auto_import	bool	If True, when a '_type' is not registered, PyGX will identify its parent module and automatically import it. For example, if the type is 'foo.bar.A', PyGX will try to import 'foo.bar' and find the class 'A' within the imported module.	True
convert_unknown	bool	If True, when a '_type' is not registered and cannot be imported, PyGX will create objects of: - pg.symbolic.UnknownType for unknown types; - pg.symbolic.UnknownTypedObject for objects of unknown types; - pg.symbolic.UnknownFunction for unknown functions; - pg.symbolic.UnknownMethod for unknown methods. If False, TypeError will be raised.	False
**kwargs	Any	Keyword arguments that will be passed to WireConvertible.init.	{}

Returns:

Type	Description
Any	Deserialized value.

Source code in pygx/wire/_conversion.py

def from_json(
    json_value: WireValue,
    *,
    context: WireConversionContext | None = None,
    auto_import: bool = True,
    convert_unknown: bool = False,
    **kwargs: Any,
) -> Any:
    """Deserializes a (maybe) WireConvertible value from JSON value.

    Args:
      json_value: Input JSON value.
      context: Serialization context.
      auto_import: If True, when a '_type' is not registered, PyGX will
        identify its parent module and automatically import it. For example,
        if the type is 'foo.bar.A', PyGX will try to import 'foo.bar' and
        find the class 'A' within the imported module.
      convert_unknown: If True, when a '_type' is not registered and cannot
        be imported, PyGX will create objects of:
          - `pg.symbolic.UnknownType` for unknown types;
          - `pg.symbolic.UnknownTypedObject` for objects of unknown types;
          - `pg.symbolic.UnknownFunction` for unknown functions;
          - `pg.symbolic.UnknownMethod` for unknown methods.
        If False, TypeError will be raised.
      **kwargs: Keyword arguments that will be passed to WireConvertible.__init__.

    Returns:
      Deserialized value.
    """
    if context is None and isinstance(json_value, dict):
        context_node = json_value.get(WireConvertible.CONTEXT_KEY)
        if context_node is not None:
            context = WireConversionContext.from_json(
                context_node,
                auto_import=auto_import,
                convert_unknown=convert_unknown,
                **kwargs,
            )
            json_value = json_value[WireConvertible.ROOT_VALUE_KEY]
        # Else: no `__context__` -> tree has no `__ref__` entries to resolve
        # against a context. Leave `context` as None.

    typename_resolved = kwargs.pop('_typename_resolved', False)
    if not typename_resolved:
        json_value = resolve_typenames(
            json_value, auto_import=auto_import, convert_unknown=convert_unknown
        )

    def child_from(v):
        return from_json(v, context=context, _typename_resolved=True, **kwargs)

    if isinstance(json_value, list):
        if json_value and json_value[0] == WireConvertible.TUPLE_MARKER:
            if len(json_value) < 2:
                raise ValueError(
                    f'Tuple should have at least one element '
                    f"besides '{WireConvertible.TUPLE_MARKER}'. "
                    f'Encountered: {json_value}.'
                )
            return tuple([child_from(v) for v in json_value[1:]])
        return [child_from(v) for v in json_value]
    elif isinstance(json_value, dict):
        if WireConvertible.REF_KEY in json_value:
            assert context is not None
            v = context.get_shared(json_value[WireConvertible.REF_KEY]).value
            return v
        if WireConvertible.TYPE_NAME_KEY not in json_value:
            return {k: child_from(v) for k, v in json_value.items()}
        factory_fn = json_value.pop(WireConvertible.TYPE_NAME_KEY)
        assert factory_fn is not None
        return factory_fn(json_value, context=context, **kwargs)
    return json_value

registered_types

registered_types() -> Iterable[tuple[str, type[WireConvertible]]]

Returns an iterator of registered (serialization key, class) tuples.

Source code in pygx/wire/_conversion.py

def registered_types() -> Iterable[tuple[str, type[WireConvertible]]]:
    """Returns an iterator of registered (serialization key, class) tuples."""
    return WireConvertible.registered_types()

to_json

to_json(
    value: Any,
    *,
    context: WireConversionContext | None = None,
    supports_refs: bool = True,
    **kwargs: Any
) -> Any

Serializes a (maybe) WireConvertible value into a plain Python object.

Parameters:

Name	Type	Description	Default
value	Any	value to serialize. Applicable value types are: Builtin python types: None, bool, int, float, string; WireConvertible types; List types; Tuple types; Dict types.	required
context	WireConversionContext | None	JSON conversion context.	None
supports_refs	bool	Whether the target wire format can represent shared/cyclic references. When False, a value referenced more than once raises instead of emitting a __ref__. Only honored when context is None (i.e. at the root call); inner recursive calls inherit the root context's setting.	True
**kwargs	Any	Keyword arguments to pass to value.to_json if value is WireConvertible.	{}

Returns:

Type	Description
Any	JSON value.

Source code in pygx/wire/_conversion.py

def to_json(
    value: Any,
    *,
    context: WireConversionContext | None = None,
    supports_refs: bool = True,
    **kwargs: Any,
) -> Any:
    """Serializes a (maybe) WireConvertible value into a plain Python object.

    Args:
      value: value to serialize. Applicable value types are:

        * Builtin python types: None, bool, int, float, string;
        * WireConvertible types;
        * List types;
        * Tuple types;
        * Dict types.

      context: JSON conversion context.
      supports_refs: Whether the target wire format can represent shared/cyclic
        references. When False, a value referenced more than once raises instead
        of emitting a `__ref__`. Only honored when `context` is None (i.e. at the
        root call); inner recursive calls inherit the root context's setting.
      **kwargs: Keyword arguments to pass to value.to_json if value is
        WireConvertible.

    Returns:
      JSON value.
    """
    # Primitive fast path: skip context allocation entirely for the common
    # case where `to_json` is called on a scalar (e.g. from a leaf
    # recursion in a Dict/List sym_jsonify).
    if value.__class__ in _PRIMITIVE_WIRE_TYPES:
        return value

    if context is None:
        is_root = True
        context = WireConversionContext(supports_refs=supports_refs)
    else:
        is_root = False

    # `sym_jsonify` is the fast type-marker for a symbolic `WireConvertible`
    # (`Object`/`Dict`/`List`/`Ref`) — present iff the value is `Symbolic`.
    # Probing it first lets the hot symbolic path skip the ABCMeta
    # `isinstance(value, WireConvertible)` check; only non-symbolic
    # `WireConvertible` (e.g. `_OpaqueObject`, `MissingValue`, user types)
    # pays it.
    sym_fn = getattr(value, 'sym_jsonify', None)
    if sym_fn is not None:
        if isinstance(value, list):
            # Symbolic `pg.List`: `sym_jsonify` returns a `list`, which
            # `serialize_maybe_shared` can't mutate in place into a
            # `{REF_KEY: idx}` if a second reference appears later. Routing
            # it through SMS would force the post-pass to walk the entire
            # output tree just to unwrap a wrap we created here. `pg.List`'s
            # single-parent invariant makes sharing within the symbolic tree
            # impossible, so inlining directly is safe in practice. (Edge:
            # multiple `Ref`s to the same `pg.List` instance would lose
            # identity dedup on roundtrip.)
            v = sym_fn(context=context, **kwargs)
        else:
            # Symbolic `Object`/`Dict`/`Ref`: SMS handles shared tracking.
            v = context.serialize_maybe_shared(value, json_fn=sym_fn, **kwargs)
    elif isinstance(value, WireConvertible):
        # Non-symbolic WireConvertible (e.g. `_OpaqueObject`, `MissingValue`,
        # user types) — no `sym_jsonify`, serialize via `to_json`.
        v = context.serialize_maybe_shared(
            value, json_fn=value.to_json, **kwargs
        )
    elif isinstance(value, list):
        # Standard lists serialize by references.
        v = context.serialize_maybe_shared(
            value,
            json_fn=lambda **kwargs: [to_json(x, **kwargs) for x in value],
            **kwargs,
        )
    elif isinstance(value, dict):
        # Standard dicts serialize by references. `exclude_none` (#519)
        # filters here too, so a RAW dict (a `sym=False` / `validate=False`
        # field value) emits the same bytes as its `pg.Dict` twin — the
        # sym-transparency law. The `v is not None` short-circuit keeps the
        # flag read off non-None entries.
        v = context.serialize_maybe_shared(
            value,
            json_fn=lambda **kwargs: {
                k: to_json(v, **kwargs)
                for k, v in value.items()  # pytype: disable=attribute-error
                if v is not None or not kwargs.get('exclude_none', False)
            },
            **kwargs,
        )
    elif isinstance(value, tuple):
        # Tuples serialize by values.
        v = [WireConvertible.TUPLE_MARKER] + [
            to_json(item, context=context, **kwargs) for item in value
        ]
    elif isinstance(value, (type, types.GenericAlias)):
        v = _type_to_json(value)
    elif inspect.isbuiltin(value):
        v = _builtin_function_to_json(value)
    elif inspect.isfunction(value):
        v = _function_to_json(value)
    elif inspect.ismethod(value):
        v = _method_to_json(value)
    # pytype: disable=module-attr
    elif isinstance(value, typing._Final):  # pylint: disable=protected-access
        # pytype: enable=module-attr
        v = _annotation_to_json(value)
    elif value is ...:
        v = {WireConvertible.TYPE_NAME_KEY: 'type', 'name': 'builtins.Ellipsis'}
    else:
        # Opaque objects serialize by references. (Types that want a non-opaque
        # wire form subclass `WireConvertible` and are handled above.)
        v = context.serialize_maybe_shared(
            value,
            json_fn=lambda **kwargs: _OpaqueObject(value).to_json(**kwargs),
            **kwargs,
        )

    if is_root:
        return context.to_json(root=v, **kwargs)
    return v

deserialize

deserialize(
    data: str | bytes, *, format: str | WireFormat = "json", **kwargs: Any
) -> Any

Deserializes a value from a wire format (JSON by default).

Parameters:

Name	Type	Description	Default
data	str | bytes	The serialized representation.	required
format	str | WireFormat	A registered format name or a :class:WireFormat instance.	'json'
**kwargs	Any	Additional keyword arguments forwarded to :func:from_json.	{}

Returns:

Type	Description
Any	The deserialized value.

Source code in pygx/wire/_format.py

def deserialize(
    data: str | bytes,
    *,
    format: str | WireFormat = 'json',  # pylint: disable=redefined-builtin
    **kwargs: Any,
) -> Any:
    """Deserializes a value from a wire format (JSON by default).

    Args:
      data: The serialized representation.
      format: A registered format name or a :class:`WireFormat` instance.
      **kwargs: Additional keyword arguments forwarded to :func:`from_json`.

    Returns:
      The deserialized value.
    """
    fmt = get_format(format)
    return from_json(fmt.decode(data), **kwargs)

get_format

get_format(format: str | WireFormat) -> WireFormat

Resolves a format name or instance to a :class:WireFormat.

Parameters:

Name	Type	Description	Default
format	str | WireFormat	Either a registered format name (e.g. 'json') or a :class:WireFormat instance (returned as-is).	required

Returns:

Type	Description
WireFormat	The resolved :class:WireFormat.

Raises:

Type	Description
KeyError	If a name is given that is not registered.

Source code in pygx/wire/_format.py

def get_format(format: str | WireFormat) -> WireFormat:  # pylint: disable=redefined-builtin
    """Resolves a format name or instance to a :class:`WireFormat`.

    Args:
      format: Either a registered format name (e.g. ``'json'``) or a
        :class:`WireFormat` instance (returned as-is).

    Returns:
      The resolved :class:`WireFormat`.

    Raises:
      KeyError: If a name is given that is not registered.
    """
    if isinstance(format, WireFormat):
        return format
    fmt = _FORMAT_REGISTRY.get(format)
    if fmt is None:
        raise KeyError(
            f'Unknown wire format {format!r}. Registered formats: '
            f'{sorted(_FORMAT_REGISTRY)}.'
        )
    return fmt

register_format

register_format(fmt: WireFormat, *, override_existing: bool = False) -> None

Registers a wire format under its name for lookup by format=.

Parameters:

Name	Type	Description	Default
fmt	WireFormat	The format instance to register.	required
override_existing	bool	If True, replace an already-registered format with the same name. Otherwise a KeyError is raised on a name clash.	False

Raises:

Type	Description
KeyError	If fmt.name is already registered and override_existing is False.

Source code in pygx/wire/_format.py

def register_format(
    fmt: WireFormat, *, override_existing: bool = False
) -> None:
    """Registers a wire format under its ``name`` for lookup by ``format=``.

    Args:
      fmt: The format instance to register.
      override_existing: If True, replace an already-registered format with the
        same name. Otherwise a ``KeyError`` is raised on a name clash.

    Raises:
      KeyError: If ``fmt.name`` is already registered and ``override_existing``
        is False.
    """
    if fmt.name in _FORMAT_REGISTRY and not override_existing:
        raise KeyError(
            f'Wire format {fmt.name!r} is already registered with '
            f'{type(_FORMAT_REGISTRY[fmt.name]).__name__}.'
        )
    _FORMAT_REGISTRY[fmt.name] = fmt

registered_formats

registered_formats() -> dict[str, WireFormat]

Returns a copy of the registered name -> WireFormat mapping.

Source code in pygx/wire/_format.py

def registered_formats() -> dict[str, WireFormat]:
    """Returns a copy of the registered ``name -> WireFormat`` mapping."""
    return dict(_FORMAT_REGISTRY)

serialize

serialize(
    value: Any,
    *,
    format: str | WireFormat = "json",
    indent: int | None = None,
    **kwargs: Any
) -> str | bytes

Serializes a value to a wire format (JSON by default).

Convenience wrapper that runs the conversion engine then the format codec::

s = pg.serialize(obj, format='yaml') obj2 = pg.deserialize(s, format='yaml')

Parameters:

Name	Type	Description	Default
value	Any	The value to serialize.	required
format	str | WireFormat	A registered format name (e.g. 'json', 'yaml') or a :class:WireFormat instance.	'json'
indent	int | None	Optional indentation passed to the format's encoder.	None
**kwargs	Any	Additional keyword arguments forwarded to :func:to_json.	{}

Returns:

Type	Description
str | bytes	The serialized representation (str or bytes, format-dependent).

Source code in pygx/wire/_format.py

def serialize(
    value: Any,
    *,
    format: str | WireFormat = 'json',  # pylint: disable=redefined-builtin
    indent: int | None = None,
    **kwargs: Any,
) -> str | bytes:
    """Serializes a value to a wire format (JSON by default).

    Convenience wrapper that runs the conversion engine then the format codec::

      s = pg.serialize(obj, format='yaml')
      obj2 = pg.deserialize(s, format='yaml')

    Args:
      value: The value to serialize.
      format: A registered format name (e.g. ``'json'``, ``'yaml'``) or a
        :class:`WireFormat` instance.
      indent: Optional indentation passed to the format's encoder.
      **kwargs: Additional keyword arguments forwarded to :func:`to_json`.

    Returns:
      The serialized representation (``str`` or ``bytes``, format-dependent).
    """
    fmt = get_format(format)
    tree = to_json(value, supports_refs=fmt.supports_refs, **kwargs)
    return fmt.encode(tree, indent=indent)

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2