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
    # Read side: stand-ins handed out for forward refs during pass 1,
    # patched by `_resolve_pending_refs` once the table is complete. Only
    # a CYCLE produces one.
    self._pending_refs: list[_PendingRef] = []

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)

pending_ref

pending_ref(ref_index: int) -> _PendingRef

A stand-in for a not-yet-built shared entry (a cycle's back-edge).

Source code in pygx/wire/_conversion.py

def pending_ref(self, ref_index: int) -> _PendingRef:
    """A stand-in for a not-yet-built shared entry (a cycle's back-edge)."""
    marker = _PendingRef(ref_index)
    self._pending_refs.append(marker)
    return marker

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 ------------------------------------------------
    # RESERVE the entry BEFORE serializing, so a value that contains
    # ITSELF finds this reservation when it re-enters and emits a
    # `{REF_KEY: idx}` back-edge instead of descending forever. (The same
    # reason `_clone_cyclic` registers its shell before filling it: an
    # entry written after the descent is written too late to stop one.)
    entry = self.ObjectEntry(
        value=value,
        serialized=None,
        ref_index=len(self._shared_objects),
        ref_count=1,
        inlined_at=None,
    )
    self._shared_objects.append(entry)
    self._id_to_shared_object[value_id] = entry
    serialized = json_fn(context=self, **kwargs)
    entry.serialized = serialized

    if entry.ref_count > 1:
        # The value referenced ITSELF while serializing — a CYCLE. The
        # back-edge already names this index, so the content must stay in
        # the shared table; inlining it at the caller would leave the
        # back-edge pointing at an entry that no longer holds anything.
        self._needs_post_pass = True
        return {WireConvertible.REF_KEY: entry.ref_index}

    # Dict-shaped serializations stay inline at the caller's location
    # (`inlined_at` points to the same dict), so the common no-share case
    # emits no ref at all. Other shapes get a `{REF_KEY: idx}` wrap that
    # the post-pass unwraps when `ref_count` stays 1.
    if isinstance(serialized, dict):
        entry.inlined_at = serialized
        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,
                    )
                )
            # Pass 2: a CYCLE references forward, so those refs stood in
            # as markers above; patch them now that the table is whole.
            # A no-op for acyclic payloads (nothing stood in).
            context._resolve_pending_refs()  # pylint: disable=protected-access
    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.

escape_wire_key

escape_wire_key(k: Any) -> Any

The emit-side escape: one added backslash on a marker-like key.

Source code in pygx/wire/_conversion.py

def escape_wire_key(k: Any) -> Any:
    """The emit-side escape: one added backslash on a marker-like key."""
    if k.__class__ is str and (k[:1] == '_' or k[:1] == '\\'):
        if k in _ESCAPABLE_KEYS or k.lstrip('\\') in _ESCAPABLE_KEYS:
            return '\\' + k
    return k

escape_wire_list_head

escape_wire_list_head(out: list) -> list

Escapes a serialized LIST whose head spells the tuple marker.

Source code in pygx/wire/_conversion.py

def escape_wire_list_head(out: list) -> list:
    """Escapes a serialized LIST whose head spells the tuple marker."""
    if out:
        h = out[0]
        if h.__class__ is str and (h[:1] == '_' or h[:1] == '\\'):
            if h in _ESCAPABLE_HEADS or h.lstrip('\\') in _ESCAPABLE_HEADS:
                out[0] = '\\' + h
    return out

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. Two names are consumed by the engine itself rather than forwarded: trusted_pickle_types (read by _OpaqueObject.from_json to override the global pickle allow-list for this call), and the private _typename_resolved, which marks the type-name pre-walk as already done for this subtree. The latter is POPPED here before **kwargs reaches any user from_json (L1322), so it never leaks into user code -- keep that pop ahead of every forward. pg.Symbolic.from_json pops the same flag independently for its own fast path.	{}

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__.
        Two names are consumed by the engine itself rather than forwarded:
        ``trusted_pickle_types`` (read by ``_OpaqueObject.from_json`` to
        override the global pickle allow-list for this call), and the private
        ``_typename_resolved``, which marks the type-name pre-walk as already
        done for this subtree. The latter is POPPED here before ``**kwargs``
        reaches any user ``from_json`` (L1322), so it never leaks into user
        code -- keep that pop ahead of every forward. ``pg.Symbolic.from_json``
        pops the same flag independently for its own fast path.

    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 (h := json_value[0]).__class__ is str
            and (h[:1] == '\\' and h.lstrip('\\') in _ESCAPABLE_HEADS)
        ):
            # An escaped literal head: strip one backslash, keep it a list.
            return [h[1:]] + [child_from(v) for v in json_value[1:]]
        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
            ref_index = json_value[WireConvertible.REF_KEY]
            # A FORWARD ref names an entry not built yet — only a CYCLE
            # produces one, since the table is written bottom-up. Stand in
            # for it; `_resolve_pending_refs` patches once all entries exist.
            if ref_index >= context.next_shared_index():
                return context.pending_ref(ref_index)
            return context.get_shared(ref_index).value
        if WireConvertible.TYPE_NAME_KEY not in json_value:
            return {
                unescape_wire_key(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
        if any(k.__class__ is str and k[:1] == '\\' for k in json_value):
            # Escaped residue keys in a typed payload (a extra field named
            # after a marker) unescape before the factory sees them.
            json_value = {
                unescape_wire_key(k): v for k, v in json_value.items()
            }
        return factory_fn(json_value, context=context, **kwargs)
    return json_value

is_wire_marker_key

is_wire_marker_key(k: Any) -> bool

True if k is a reserved wire marker key, unescaped.

The DECLARATION-side test (schema application rejects a symbolic field spelled this way), as opposed to escape_wire_key's emit-side one: this asks "is this the marker itself", so an already-escaped spelling (\_type) is NOT a marker — it round-trips to the literal key.

Source code in pygx/wire/_conversion.py

def is_wire_marker_key(k: Any) -> bool:
    """True if `k` is a reserved wire marker key, unescaped.

    The DECLARATION-side test (schema application rejects a symbolic field
    spelled this way), as opposed to `escape_wire_key`'s emit-side one:
    this asks "is this the marker itself", so an already-escaped spelling
    (`\\_type`) is NOT a marker — it round-trips to the literal key.
    """
    return k.__class__ is str and k in _ESCAPABLE_KEYS

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.
    """
    # DISPATCH ORDER IS LOAD-BEARING -- the arms below are not mutually
    # exclusive, so each must precede every arm it could also match:
    #
    # * the ``sym_jsonify`` probe precedes ``isinstance(value, list)`` and
    #   ``dict``, because a ``pg.List`` IS a ``list`` and a ``pg.Dict`` IS a
    #   ``dict``; reaching the raw-container arm first would drop symbolic
    #   framing.
    # * the ``WireConvertible`` arm precedes them for the same reason (a user
    #   subclass may also subclass ``list``/``dict``).
    # * ``_STDLIB_SCALAR_ENCODERS`` sits just above the opaque fallback so a
    #   SUBCLASS of a tagged scalar keeps opaque-pickle semantics (#655).
    #
    # These are ``isinstance`` tests, not exact-type tests, ON PURPOSE:
    # ``list``/``dict``/``tuple`` subclasses must serialize as their builtin
    # shape, and ``WireConvertible`` / ``typing._Final`` are abstract. A
    # ``type()``-keyed registry would silently route every such subclass to
    # the opaque-pickle fallback -- a wire-format regression, not a speedup.
    # New wire types extend the chain at the position their overlap dictates.
    #
    # 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: escape_wire_list_head(
                [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 `topo=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: {
                escape_wire_key(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'}
    elif (codec := _STDLIB_SCALAR_ENCODERS.get(value.__class__)) is not None:
        # Stdlib scalar (#655): tagged canonical string, emitted BY VALUE
        # (immutable scalars — no ref-sharing bookkeeping), just before the
        # opaque fallback so subclasses keep opaque-pickle semantics.
        tag, encode = codec
        v = {WireConvertible.TYPE_NAME_KEY: tag, 'value': encode(value)}
    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

unescape_wire_key

unescape_wire_key(k: Any) -> Any

The read-side unescape: strips exactly one backslash.

Source code in pygx/wire/_conversion.py

def unescape_wire_key(k: Any) -> Any:
    """The read-side unescape: strips exactly one backslash."""
    if (
        k.__class__ is str
        and k[:1] == '\\'
        and k.lstrip('\\') in _ESCAPABLE_KEYS
    ):
        return k[1:]
    return k

unescape_wire_list_head

unescape_wire_list_head(out: list) -> list

The read-side twin of escape_wire_list_head — strips one backslash.

Mutates and returns out: every caller owns the list it passes (a freshly decoded payload), and the escape twin above mutates too.

Source code in pygx/wire/_conversion.py

def unescape_wire_list_head(out: list) -> list:
    """The read-side twin of `escape_wire_list_head` — strips one backslash.

    Mutates and returns `out`: every caller owns the list it passes (a
    freshly decoded payload), and the escape twin above mutates too.
    """
    if out:
        h = out[0]
        if (
            h.__class__ is str
            and h[:1] == '\\'
            and h.lstrip('\\') in _ESCAPABLE_HEADS
        ):
            out[0] = h[1:]
    return out

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