Python API reference

This is the complete, automatically generated Python API reference for omnist, generated directly from codebase docstrings using mkdocstrings.

Looking for a conceptual introduction and summary tables? See the API overview.

---

Public Surface (omnist)

omnist — one canonical data model, many formats.

A Document is a tree: an ordered list of labeled edges (repeated labels are how arrays appear), held by a :class:Doc. A Schema describes the shape a Document may have, as record definitions referenced by name. A field's value side is always exactly one of the seven scalars (string, integer, number, boolean, date, time, datetime), optionally nullable, or a reference to a named record — never a composed value-domain (no enums, no literal-valued fields). Read a format into a Doc, validate it against a Schema, and write it back to any format.

from omnist import parse_schema, doc

s = parse_schema('''
    record Member { "name": string, "role": string }
    record Team   { "name": string, "members" [1,]: Member }
    root Team
''')
s.validate(doc({"name": "X", "members": [{"name": "Ann", "role": "dev"}]})).ok

The model is defined formally in docs/design/model.md; this package is its implementation, and this module is its public surface.

t = _Types() module-attribute

STRING = Scalar('string') module-attribute

INTEGER = Scalar('integer') module-attribute

NUMBER = Scalar('number') module-attribute

BOOLEAN = Scalar('boolean') module-attribute

DATE = Scalar('date') module-attribute

TIME = Scalar('time') module-attribute

DATETIME = Scalar('datetime') module-attribute

Doc

A guarded handle on a Document node (a leaf value or an edge list).

Source code in omnist/document.py

class Doc:
    """A guarded handle on a Document node (a leaf value or an edge list)."""

    __slots__ = ("_node", "path", "depth")

    def __init__(self, node: Any, path: str = "$", depth: int = 0) -> None:
        """Initialize a Doc cursor pointing to ``node`` at ``path``."""
        self._node = node
        self.path = path
        self.depth = depth

    # -- construction ---------------------------------------------------
    @classmethod
    def of(cls, value: Any) -> "Doc":
        """Construct a Doc from a plain Python value (JSON-shaped mapping, sequence, or scalar)."""
        return cls(build_node(value))

    @classmethod
    def from_format(cls, name: str, text: str) -> "Doc":
        """Parse source text using the registered format named ``name`` into a Doc."""
        from .registry import get_format
        return cls(get_format(name).read(text))

    @classmethod
    def from_json(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse JSON text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_json
        return cls(read_json(text, schema=schema))

    @classmethod
    def from_yaml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse YAML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_yaml
        return cls(read_yaml(text, schema=schema))

    @classmethod
    def from_toml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse TOML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_toml
        return cls(read_toml(text, schema=schema))

    @classmethod
    def from_xml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse XML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_xml
        return cls(read_xml(text, schema=schema))

    @classmethod
    def from_oml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse OML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .oml import read_oml
        return cls(read_oml(text, schema=schema))

    # -- shape ----------------------------------------------------------
    @property
    def is_leaf(self) -> bool:
        """True if this cursor points to a scalar leaf value (spec §2.1)."""
        return not isinstance(self._node, list)

    @property
    def value(self) -> Any:
        """The scalar value at this leaf node (spec §2.2).

        Raises :class:`~omnist.errors.DocumentError` if this is an internal node.
        """
        if isinstance(self._node, list):
            raise DocumentError(f"{self.path}: not a leaf; use edges()")
        return self._node

    def edges(self) -> List[Tuple[str, "Doc"]]:
        """Return the ordered list of ``(label, Doc)`` child edges (spec §2.1).

        Raises :class:`~omnist.errors.DocumentError` if this is a leaf node.
        """
        if not isinstance(self._node, list):
            raise DocumentError(f"{self.path}: a leaf has no edges")
        out: List[Tuple[str, "Doc"]] = []
        counts: dict[str, int] = {}
        for label, child in self._node:
            i = counts.get(label, 0)
            counts[label] = i + 1
            cp = f"{self.path}.{label}" if i == 0 else f"{self.path}.{label}[{i}]"
            out.append((label, Doc(child, cp, self.depth + 1)))
        return out

    def labels(self) -> List[str]:
        """Return the deduplicated list of child edge labels in first-occurrence order."""
        seen: set[str] = set()
        out: List[str] = []
        for label, _ in self._iter():
            if label not in seen:
                seen.add(label)
                out.append(label)
        return out

    def get(self, label: str) -> List["Doc"]:
        """Return all child Doc cursors matching ``label``."""
        return [c for lbl, c in self.edges() if lbl == label]

    def get_one(self, label: str) -> "Doc":
        """Return the single child Doc cursor matching ``label``.

        Raises :class:`~omnist.errors.DocumentError` if there are not exactly 1 matching edge.
        """
        cs = self.get(label)
        if len(cs) != 1:
            raise DocumentError(
                f"{self.path}: expected exactly one {label!r}, found {len(cs)}")
        return cs[0]

    def count(self, label: str) -> int:
        """Return the number of child edges matching ``label``."""
        return sum(1 for lbl, _ in self._iter() if lbl == label)

    def _iter(self) -> Iterator[Tuple[str, Any]]:
        if isinstance(self._node, list):
            yield from self._node

    def child(self, label: str) -> "Doc":
        """A cursor to the single child under ``label`` (editable if internal)."""
        return self.get_one(label)

    # -- editing (mutates the underlying edge list) ---------------------
    def add(self, label: str, value: Any) -> "Doc":
        """Append an edge ``(label, value)``.  A repeated label is how an array
        grows.  Returns ``self`` for chaining."""
        self._require_internal("add")
        self._node.append(
            (label, build_node(value, f"{self.path}.{label}", self.depth + 1)))
        return self

    def remove(self, label: str) -> "Doc":
        """Remove every edge under ``label``."""
        self._require_internal("remove")
        self._node[:] = [(lbl, c) for lbl, c in self._node if lbl != label]
        return self

    def set(self, label: str, value: Any) -> "Doc":
        """Replace all edges under ``label`` with a single new edge (positioned
        at the first old occurrence); ``set`` = ``remove`` + ``add``."""
        self._require_internal("set")
        new = build_node(value, f"{self.path}.{label}", self.depth + 1)
        first = None
        kept: List[Edge] = []
        for lbl, child in self._node:
            if lbl == label:
                if first is None:
                    first = len(kept)
                    kept.append((label, new))
                # later duplicates are dropped
            else:
                kept.append((lbl, child))
        if first is None:
            kept.append((label, new))
        self._node[:] = kept
        return self

    def _require_internal(self, op: str) -> None:
        if not isinstance(self._node, list):
            raise DocumentError(f"{self.path}: cannot {op} on a leaf")

    # -- export ---------------------------------------------------------
    def to_data(self) -> Any:
        """Return a deep copy of the raw underlying node representation (spec §2.1)."""
        return _copy(self._node)

    def to_grouped(self) -> Any:
        """A JSON-shaped projection: same-label edges grouped into a list.

        A label seen once stays a single value; a label seen more than once
        becomes a list (the schema-less fallback of the count-1 rule, see
        ``docs/design/model.md`` §10)."""
        return _grouped(self._node)

    def to_json(self, **o: Any) -> str:
        """Serialize this Document to JSON text (spec §4)."""
        from .formats import write_json
        return write_json(self._node, **o)

    def to_yaml(self, **o: Any) -> str:
        """Serialize this Document to YAML text (spec §4)."""
        from .formats import write_yaml
        return write_yaml(self._node, **o)

    def to_toml(self, **o: Any) -> str:
        """Serialize this Document to TOML text (spec §4)."""
        from .formats import write_toml
        return write_toml(self._node, **o)

    def to_xml(self, **o: Any) -> str:
        """Serialize this Document to XML text (spec §4)."""
        from .formats import write_xml
        return write_xml(self._node, **o)

    def to_oml(self, **o: Any) -> str:
        """Serialize this Document to OML text (spec §4)."""
        from .oml import write_oml
        return write_oml(self._node, **o)

    def to_format(self, name: str, **o: Any) -> str:
        """Serialize this Document to the registered format named ``name``."""
        from .registry import get_format
        return get_format(name).write(self._node, **o)

    def check_json(self) -> "WriteReport":
        """Simulate writing to JSON and return the adjustment report (spec §4)."""
        from .formats import check_json
        return check_json(self._node)

    def check_yaml(self) -> "WriteReport":
        """Simulate writing to YAML and return the adjustment report (spec §4)."""
        from .formats import check_yaml
        return check_yaml(self._node)

    def check_toml(self) -> "WriteReport":
        """Simulate writing to TOML and return the adjustment report (spec §4)."""
        from .formats import check_toml
        return check_toml(self._node)

    def check_xml(self) -> "WriteReport":
        """Simulate writing to XML and return the adjustment report (spec §4)."""
        from .formats import check_xml
        return check_xml(self._node)

    def check_oml(self) -> "WriteReport":
        """Simulate writing to OML and return the adjustment report (spec §4)."""
        from .oml import check_oml
        return check_oml(self._node)

    def check_format(self, name: str) -> "WriteReport":
        """Simulate writing to format ``name`` and return the adjustment
        report, without producing output. Requires the registered
        :class:`~omnist.registry.Format` to provide a ``check``
        callable (the four built-ins do; a custom plugin may not)."""
        from .registry import get_format
        fmt = get_format(name)
        if fmt.check is None:
            raise DocumentError(
                f"format {name!r} has no check() -- cannot simulate a write")
        return fmt.check(self._node)  # type: ignore[no-any-return]

    def validate(self, schema: "Schema") -> "ValidationResult":
        """Validate this Document against ``schema`` (spec §5)."""
        return schema.validate(self)

    # -- dunders --------------------------------------------------------
    def __eq__(self, other: Any) -> bool:
        if isinstance(other, Doc):
            return self._node == other._node  # type: ignore[no-any-return]
        try:
            return self._node == build_node(other)  # type: ignore[no-any-return]
        except DocumentError:
            return NotImplemented

    def __repr__(self) -> str:
        return f"Doc({'leaf' if self.is_leaf else 'node'}: {self._node!r})"

is_leaf property

True if this cursor points to a scalar leaf value (spec §2.1).

value property

The scalar value at this leaf node (spec §2.2).

Raises :class:~omnist.errors.DocumentError if this is an internal node.

__init__(node, path='$', depth=0)

Initialize a Doc cursor pointing to node at path.

Source code in omnist/document.py

def __init__(self, node: Any, path: str = "$", depth: int = 0) -> None:
    """Initialize a Doc cursor pointing to ``node`` at ``path``."""
    self._node = node
    self.path = path
    self.depth = depth

add(label, value)

Append an edge (label, value). A repeated label is how an array grows. Returns self for chaining.

Source code in omnist/document.py

def add(self, label: str, value: Any) -> "Doc":
    """Append an edge ``(label, value)``.  A repeated label is how an array
    grows.  Returns ``self`` for chaining."""
    self._require_internal("add")
    self._node.append(
        (label, build_node(value, f"{self.path}.{label}", self.depth + 1)))
    return self

check_format(name)

Simulate writing to format name and return the adjustment report, without producing output. Requires the registered :class:~omnist.registry.Format to provide a check callable (the four built-ins do; a custom plugin may not).

Source code in omnist/document.py

def check_format(self, name: str) -> "WriteReport":
    """Simulate writing to format ``name`` and return the adjustment
    report, without producing output. Requires the registered
    :class:`~omnist.registry.Format` to provide a ``check``
    callable (the four built-ins do; a custom plugin may not)."""
    from .registry import get_format
    fmt = get_format(name)
    if fmt.check is None:
        raise DocumentError(
            f"format {name!r} has no check() -- cannot simulate a write")
    return fmt.check(self._node)  # type: ignore[no-any-return]

check_json()

Simulate writing to JSON and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_json(self) -> "WriteReport":
    """Simulate writing to JSON and return the adjustment report (spec §4)."""
    from .formats import check_json
    return check_json(self._node)

check_oml()

Simulate writing to OML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_oml(self) -> "WriteReport":
    """Simulate writing to OML and return the adjustment report (spec §4)."""
    from .oml import check_oml
    return check_oml(self._node)

check_toml()

Simulate writing to TOML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_toml(self) -> "WriteReport":
    """Simulate writing to TOML and return the adjustment report (spec §4)."""
    from .formats import check_toml
    return check_toml(self._node)

check_xml()

Simulate writing to XML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_xml(self) -> "WriteReport":
    """Simulate writing to XML and return the adjustment report (spec §4)."""
    from .formats import check_xml
    return check_xml(self._node)

check_yaml()

Simulate writing to YAML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_yaml(self) -> "WriteReport":
    """Simulate writing to YAML and return the adjustment report (spec §4)."""
    from .formats import check_yaml
    return check_yaml(self._node)

child(label)

A cursor to the single child under label (editable if internal).

Source code in omnist/document.py

def child(self, label: str) -> "Doc":
    """A cursor to the single child under ``label`` (editable if internal)."""
    return self.get_one(label)

count(label)

Return the number of child edges matching label.

Source code in omnist/document.py

def count(self, label: str) -> int:
    """Return the number of child edges matching ``label``."""
    return sum(1 for lbl, _ in self._iter() if lbl == label)

edges()

Return the ordered list of (label, Doc) child edges (spec §2.1).

Raises :class:~omnist.errors.DocumentError if this is a leaf node.

Source code in omnist/document.py

def edges(self) -> List[Tuple[str, "Doc"]]:
    """Return the ordered list of ``(label, Doc)`` child edges (spec §2.1).

    Raises :class:`~omnist.errors.DocumentError` if this is a leaf node.
    """
    if not isinstance(self._node, list):
        raise DocumentError(f"{self.path}: a leaf has no edges")
    out: List[Tuple[str, "Doc"]] = []
    counts: dict[str, int] = {}
    for label, child in self._node:
        i = counts.get(label, 0)
        counts[label] = i + 1
        cp = f"{self.path}.{label}" if i == 0 else f"{self.path}.{label}[{i}]"
        out.append((label, Doc(child, cp, self.depth + 1)))
    return out

from_format(name, text) classmethod

Parse source text using the registered format named name into a Doc.

Source code in omnist/document.py

@classmethod
def from_format(cls, name: str, text: str) -> "Doc":
    """Parse source text using the registered format named ``name`` into a Doc."""
    from .registry import get_format
    return cls(get_format(name).read(text))

from_json(text, *, schema=None) classmethod

Parse JSON text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_json(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse JSON text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_json
    return cls(read_json(text, schema=schema))

from_oml(text, *, schema=None) classmethod

Parse OML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_oml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse OML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .oml import read_oml
    return cls(read_oml(text, schema=schema))

from_toml(text, *, schema=None) classmethod

Parse TOML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_toml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse TOML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_toml
    return cls(read_toml(text, schema=schema))

from_xml(text, *, schema=None) classmethod

Parse XML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_xml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse XML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_xml
    return cls(read_xml(text, schema=schema))

from_yaml(text, *, schema=None) classmethod

Parse YAML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_yaml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse YAML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_yaml
    return cls(read_yaml(text, schema=schema))

get(label)

Return all child Doc cursors matching label.

Source code in omnist/document.py

def get(self, label: str) -> List["Doc"]:
    """Return all child Doc cursors matching ``label``."""
    return [c for lbl, c in self.edges() if lbl == label]

get_one(label)

Return the single child Doc cursor matching label.

Raises :class:~omnist.errors.DocumentError if there are not exactly 1 matching edge.

Source code in omnist/document.py

def get_one(self, label: str) -> "Doc":
    """Return the single child Doc cursor matching ``label``.

    Raises :class:`~omnist.errors.DocumentError` if there are not exactly 1 matching edge.
    """
    cs = self.get(label)
    if len(cs) != 1:
        raise DocumentError(
            f"{self.path}: expected exactly one {label!r}, found {len(cs)}")
    return cs[0]

labels()

Return the deduplicated list of child edge labels in first-occurrence order.

Source code in omnist/document.py

def labels(self) -> List[str]:
    """Return the deduplicated list of child edge labels in first-occurrence order."""
    seen: set[str] = set()
    out: List[str] = []
    for label, _ in self._iter():
        if label not in seen:
            seen.add(label)
            out.append(label)
    return out

of(value) classmethod

Construct a Doc from a plain Python value (JSON-shaped mapping, sequence, or scalar).

Source code in omnist/document.py

@classmethod
def of(cls, value: Any) -> "Doc":
    """Construct a Doc from a plain Python value (JSON-shaped mapping, sequence, or scalar)."""
    return cls(build_node(value))

remove(label)

Remove every edge under label.

Source code in omnist/document.py

def remove(self, label: str) -> "Doc":
    """Remove every edge under ``label``."""
    self._require_internal("remove")
    self._node[:] = [(lbl, c) for lbl, c in self._node if lbl != label]
    return self

set(label, value)

Replace all edges under label with a single new edge (positioned at the first old occurrence); set = remove + add.

Source code in omnist/document.py

def set(self, label: str, value: Any) -> "Doc":
    """Replace all edges under ``label`` with a single new edge (positioned
    at the first old occurrence); ``set`` = ``remove`` + ``add``."""
    self._require_internal("set")
    new = build_node(value, f"{self.path}.{label}", self.depth + 1)
    first = None
    kept: List[Edge] = []
    for lbl, child in self._node:
        if lbl == label:
            if first is None:
                first = len(kept)
                kept.append((label, new))
            # later duplicates are dropped
        else:
            kept.append((lbl, child))
    if first is None:
        kept.append((label, new))
    self._node[:] = kept
    return self

to_data()

Return a deep copy of the raw underlying node representation (spec §2.1).

Source code in omnist/document.py

def to_data(self) -> Any:
    """Return a deep copy of the raw underlying node representation (spec §2.1)."""
    return _copy(self._node)

to_format(name, **o)

Serialize this Document to the registered format named name.

Source code in omnist/document.py

def to_format(self, name: str, **o: Any) -> str:
    """Serialize this Document to the registered format named ``name``."""
    from .registry import get_format
    return get_format(name).write(self._node, **o)

to_grouped()

A JSON-shaped projection: same-label edges grouped into a list.

A label seen once stays a single value; a label seen more than once becomes a list (the schema-less fallback of the count-1 rule, see docs/design/model.md §10).

Source code in omnist/document.py

def to_grouped(self) -> Any:
    """A JSON-shaped projection: same-label edges grouped into a list.

    A label seen once stays a single value; a label seen more than once
    becomes a list (the schema-less fallback of the count-1 rule, see
    ``docs/design/model.md`` §10)."""
    return _grouped(self._node)

to_json(**o)

Serialize this Document to JSON text (spec §4).

Source code in omnist/document.py

def to_json(self, **o: Any) -> str:
    """Serialize this Document to JSON text (spec §4)."""
    from .formats import write_json
    return write_json(self._node, **o)

to_oml(**o)

Serialize this Document to OML text (spec §4).

Source code in omnist/document.py

def to_oml(self, **o: Any) -> str:
    """Serialize this Document to OML text (spec §4)."""
    from .oml import write_oml
    return write_oml(self._node, **o)

to_toml(**o)

Serialize this Document to TOML text (spec §4).

Source code in omnist/document.py

def to_toml(self, **o: Any) -> str:
    """Serialize this Document to TOML text (spec §4)."""
    from .formats import write_toml
    return write_toml(self._node, **o)

to_xml(**o)

Serialize this Document to XML text (spec §4).

Source code in omnist/document.py

def to_xml(self, **o: Any) -> str:
    """Serialize this Document to XML text (spec §4)."""
    from .formats import write_xml
    return write_xml(self._node, **o)

to_yaml(**o)

Serialize this Document to YAML text (spec §4).

Source code in omnist/document.py

def to_yaml(self, **o: Any) -> str:
    """Serialize this Document to YAML text (spec §4)."""
    from .formats import write_yaml
    return write_yaml(self._node, **o)

validate(schema)

Validate this Document against schema (spec §5).

Source code in omnist/document.py

def validate(self, schema: "Schema") -> "ValidationResult":
    """Validate this Document against ``schema`` (spec §5)."""
    return schema.validate(self)

Schema

A schema: a root reference plus an environment of named records.

Source code in omnist/schema.py

class Schema:
    """A schema: a root reference plus an environment of named records."""

    def __init__(self, root: Ref, env: Optional[Dict[str, Record]] = None) -> None:
        if not isinstance(root, Ref):
            raise SchemaError("a schema root must be a Ref to a named record")
        self.root = root
        self.env: Dict[str, Record] = dict(env or {})
        self.check_refs()

    def resolve(self, t: Type) -> Union[Record, Scalar, AnyType]:
        """An ``AnyType`` or bare ``Scalar`` resolves to itself; a ``Ref`` is a
        single environment lookup -- env values are always Records (enforced
        by ``check_refs``), so ref chains cannot occur."""
        if isinstance(t, AnyType):
            return t
        if isinstance(t, Scalar):
            return t
        if t.name not in self.env:
            raise SchemaError(f"unknown type {t.name!r}", code="schema.unknown-type")
        return self.env[t.name]

    def check_refs(self) -> None:
        for name, rec in self.env.items():
            if not isinstance(rec, Record):
                raise SchemaError(
                    f"environment entry {name!r} must be a Record, got {rec!r}")
            if name in SCALAR_NAMES or name == "any":
                raise SchemaError(
                    f"record name {name!r} shadows a scalar keyword; type "
                    "position resolves a bare name to a builtin first, so "
                    "this record could never be referenced",
                    code="schema.reserved-name", path=name)

        def walk(t: Type, path: str) -> None:
            if isinstance(t, Ref) and t.name not in self.env:
                # E-13: a Schema path -- the field that names the missing
                # record, or `$` when it is the root declaration itself.
                raise SchemaError(f"unknown type {t.name!r}",
                                  code="schema.unknown-type", path=path)
        walk(self.root, "$")
        for name, rec in self.env.items():
            for f in rec.fields:
                walk(f.type, f"{name}.{f.label}")
        # every env value is now known to be a Record (checked above), and
        # root is always a Ref (checked in __init__), so resolve(root) always
        # lands on a Record once the walk above confirms root.name is known.

    # -- validation -----------------------------------------------------
    def validate(self, doc: Any) -> ValidationResult:
        from .document import Doc
        if not isinstance(doc, Doc):
            raise TypeError("validate() expects a Doc; wrap your data with doc(...)")
        res = ValidationResult()
        self._conform(doc, self.root, res, 0)
        return res

    def accepts(self, doc: Any) -> bool:
        return self.validate(doc).ok

    def _conform(self, doc: Any, t: Type, res: ValidationResult, depth: int) -> None:
        from .document import _MAX_DEPTH
        if depth > _MAX_DEPTH:
            raise DocumentError(
                f"{doc.path}: nesting exceeds the maximum depth ({_MAX_DEPTH})")
        d = self.resolve(t)
        if isinstance(d, AnyType):
            return
        if isinstance(d, Scalar):
            self._conform_scalar(doc, d, res)
        else:
            self._conform_record(doc, d, res, depth)

    def _conform_scalar(self, doc: Any, s: Scalar, res: ValidationResult) -> None:
        if not doc.is_leaf:
            res.add(
                doc.path, f"expected a {s.name} value, got an object", "validate.shape-mismatch"
            )
            return
        v = doc.value
        if v is None:
            if not s.nullable:
                res.add(doc.path, "null not allowed here", "validate.null-not-allowed")
            return
        if not matches_kind(v, s.name):
            res.add(
                doc.path, f"expected {s.name}, got {_typename(v)} ({v!r})", "validate.type-mismatch"
            )

    def _conform_record(self, doc: Any, rec: Record, res: ValidationResult,
                        depth: int) -> None:
        if doc.is_leaf:
            res.add(doc.path, "expected an object, got a value", "validate.shape-mismatch")
            return
        counts: Dict[str, int] = {}
        for label, child in doc.edges():
            counts[label] = counts.get(label, 0) + 1
            f = rec.field(label)
            if f is None:
                res.add(child.path, "unexpected field", "validate.unexpected-field")
            else:
                self._conform(child, f.type, res, depth + 1)
        for f in rec.fields:
            c = counts.get(f.label, 0)
            if c < f.min or (f.max is not None and c > f.max):
                res.add(doc.path,
                        f"field {f.label!r} occurs {c} time(s), expected {f.cardinality_str()}",
                        "validate.cardinality")

    # -- comparison (delegate to operations) ----------------------------
    def compatible_with(self, other: "Schema") -> bool:
        """True if every document this schema accepts is also accepted by
        ``other`` (this is a subschema; ``other`` is backward-compatible)."""
        from .ops import compatible_with
        return compatible_with(self, other)

    def equivalent(self, other: "Schema") -> bool:
        """True if both schemas accept exactly the same documents."""
        from .ops import equivalent
        return equivalent(self, other)

    def isomorphic_to(self, other: "Schema") -> bool:
        """Stricter than ``equivalent()`` -- same document language is
        necessary but not sufficient; this additionally requires the same
        record graph structure, up to a renaming of records. Not a
        replacement for ``equivalent()`` as the definition of schema
        equality; used where a caller specifically needs to detect
        structural differences that don't change accepted documents (e.g.
        ``omnist-spec``'s conformance harness checking ``infer``'s output,
        since ``infer`` is documented to never merge duplicate records the
        way ``normalize`` does)."""
        from .ops.isomorphic import _isomorphic
        return _isomorphic(self, other)

    def normalize(self) -> "Schema":
        """The canonical minimal schema equivalent to this one: fewest env
        records, unique up to naming (via partition refinement)."""
        from .ops import normalize
        return normalize(self)

    def is_empty(self) -> bool:
        """True iff this schema's root record is unsatisfiable -- no finite
        document conforms to it (the schema's language is empty)."""
        from .ops import is_empty
        return is_empty(self)

    def prune(self) -> "Schema":
        """An equivalent schema with everything that can never match
        removed: unreachable records, never-emittable (``max == 0``)
        fields, and optional fields whose type can never be satisfied."""
        from .ops import prune
        return prune(self)

    def extract(self, *labels: str) -> "Schema":
        """The minimal subschema that only recognizes documents built from
        ``labels`` (paper Algorithm 5, ExtractSubschema). Fields whose label
        isn't in ``labels`` are dropped; if that deletes a mandatory
        (``min >= 1``) field, the record that had it -- and transitively
        anything that mandatorily depends on it -- is invalidated. Raises
        :class:`~omnist.SchemaError` if the root itself ends up invalidated
        (no valid subschema exists for this label set)."""
        from .ops import extract
        return extract(self, labels)

    # -- serialization --------------------------------------------------
    def to_osd(self, *, indent: Optional[int] = 4) -> str:
        from .osd import to_osd
        return to_osd(self, indent=indent)

    def __repr__(self) -> str:
        return f"Schema(root={self.root!r}, env={list(self.env)})"

    def __eq__(self, other: Any) -> bool:
        # env is a plain dict; dict equality is already order-independent
        # (compares keys and values, ignores insertion order), and
        # declaration order is preserved only for OSD-text readability,
        # not semantically significant -- see Record.__eq__ above.
        return (isinstance(other, Schema) and self.root == other.root
                and self.env == other.env)

compatible_with(other)

True if every document this schema accepts is also accepted by other (this is a subschema; other is backward-compatible).

Source code in omnist/schema.py

def compatible_with(self, other: "Schema") -> bool:
    """True if every document this schema accepts is also accepted by
    ``other`` (this is a subschema; ``other`` is backward-compatible)."""
    from .ops import compatible_with
    return compatible_with(self, other)

equivalent(other)

True if both schemas accept exactly the same documents.

Source code in omnist/schema.py

def equivalent(self, other: "Schema") -> bool:
    """True if both schemas accept exactly the same documents."""
    from .ops import equivalent
    return equivalent(self, other)

extract(*labels)

The minimal subschema that only recognizes documents built from labels (paper Algorithm 5, ExtractSubschema). Fields whose label isn't in labels are dropped; if that deletes a mandatory (min >= 1) field, the record that had it -- and transitively anything that mandatorily depends on it -- is invalidated. Raises :class:~omnist.SchemaError if the root itself ends up invalidated (no valid subschema exists for this label set).

Source code in omnist/schema.py

def extract(self, *labels: str) -> "Schema":
    """The minimal subschema that only recognizes documents built from
    ``labels`` (paper Algorithm 5, ExtractSubschema). Fields whose label
    isn't in ``labels`` are dropped; if that deletes a mandatory
    (``min >= 1``) field, the record that had it -- and transitively
    anything that mandatorily depends on it -- is invalidated. Raises
    :class:`~omnist.SchemaError` if the root itself ends up invalidated
    (no valid subschema exists for this label set)."""
    from .ops import extract
    return extract(self, labels)

is_empty()

True iff this schema's root record is unsatisfiable -- no finite document conforms to it (the schema's language is empty).

Source code in omnist/schema.py

def is_empty(self) -> bool:
    """True iff this schema's root record is unsatisfiable -- no finite
    document conforms to it (the schema's language is empty)."""
    from .ops import is_empty
    return is_empty(self)

isomorphic_to(other)

Stricter than equivalent() -- same document language is necessary but not sufficient; this additionally requires the same record graph structure, up to a renaming of records. Not a replacement for equivalent() as the definition of schema equality; used where a caller specifically needs to detect structural differences that don't change accepted documents (e.g. omnist-spec's conformance harness checking infer's output, since infer is documented to never merge duplicate records the way normalize does).

Source code in omnist/schema.py

def isomorphic_to(self, other: "Schema") -> bool:
    """Stricter than ``equivalent()`` -- same document language is
    necessary but not sufficient; this additionally requires the same
    record graph structure, up to a renaming of records. Not a
    replacement for ``equivalent()`` as the definition of schema
    equality; used where a caller specifically needs to detect
    structural differences that don't change accepted documents (e.g.
    ``omnist-spec``'s conformance harness checking ``infer``'s output,
    since ``infer`` is documented to never merge duplicate records the
    way ``normalize`` does)."""
    from .ops.isomorphic import _isomorphic
    return _isomorphic(self, other)

normalize()

The canonical minimal schema equivalent to this one: fewest env records, unique up to naming (via partition refinement).

Source code in omnist/schema.py

def normalize(self) -> "Schema":
    """The canonical minimal schema equivalent to this one: fewest env
    records, unique up to naming (via partition refinement)."""
    from .ops import normalize
    return normalize(self)

prune()

An equivalent schema with everything that can never match removed: unreachable records, never-emittable (max == 0) fields, and optional fields whose type can never be satisfied.

Source code in omnist/schema.py

def prune(self) -> "Schema":
    """An equivalent schema with everything that can never match
    removed: unreachable records, never-emittable (``max == 0``)
    fields, and optional fields whose type can never be satisfied."""
    from .ops import prune
    return prune(self)

resolve(t)

An AnyType or bare Scalar resolves to itself; a Ref is a single environment lookup -- env values are always Records (enforced by check_refs), so ref chains cannot occur.

Source code in omnist/schema.py

def resolve(self, t: Type) -> Union[Record, Scalar, AnyType]:
    """An ``AnyType`` or bare ``Scalar`` resolves to itself; a ``Ref`` is a
    single environment lookup -- env values are always Records (enforced
    by ``check_refs``), so ref chains cannot occur."""
    if isinstance(t, AnyType):
        return t
    if isinstance(t, Scalar):
        return t
    if t.name not in self.env:
        raise SchemaError(f"unknown type {t.name!r}", code="schema.unknown-type")
    return self.env[t.name]

Record

A closed set of named fields (constrained by its child labels).

Source code in omnist/schema.py

class Record:
    """A closed set of named fields (constrained by its child labels)."""

    __slots__ = ("fields", "_by_label")

    def __init__(self, fields: List[Field]) -> None:
        self.fields = list(fields)
        self._by_label: Dict[str, Field] = {}
        seen = set()
        for f in self.fields:
            if f.label in seen:
                raise SchemaError(f"duplicate field label {f.label!r} in a record",
                                  code="schema.duplicate-field")
            seen.add(f.label)
            self._by_label[f.label] = f

    def field(self, label: str) -> Optional[Field]:
        return self._by_label.get(label)

    def __repr__(self) -> str:
        return "record{" + ", ".join(repr(f) for f in self.fields) + "}"

    def __eq__(self, other: Any) -> bool:
        # Fields form an unordered set at the model layer (declaration
        # order isn't semantically significant) -- comparing the
        # label-keyed dicts is order-independent for free (dict equality
        # ignores insertion order), and duplicate labels are already
        # rejected by __init__, so this dict is exactly the field set.
        return isinstance(other, Record) and self._by_label == other._by_label

Scalar

One of the seven predefined value types, optionally nullable.

STRING, INTEGER, … (also under t.*) are ready-to-use, non-nullable instances — a field's type can be one of them directly, with no wrapping needed. Use :func:nullable for the ? form.

Source code in omnist/schema.py

class Scalar:
    """One of the seven predefined value types, optionally nullable.

    ``STRING``, ``INTEGER``, … (also under ``t.*``) are ready-to-use,
    non-nullable instances — a field's type can be one of them directly,
    with no wrapping needed.  Use :func:`nullable` for the ``?`` form.
    """
    __slots__ = ("name", "nullable")

    def __init__(self, name: str, nullable: bool = False) -> None:
        if name not in SCALAR_NAMES:
            raise SchemaError(f"unknown scalar {name!r}; expected one of "
                              f"{sorted(SCALAR_NAMES)}")
        self.name = name
        self.nullable = bool(nullable)

    def __repr__(self) -> str:
        return f"{self.name}{'?' if self.nullable else ''}"

    def __eq__(self, other: Any) -> bool:
        return (isinstance(other, Scalar) and other.name == self.name
                and other.nullable == self.nullable)

    def __hash__(self) -> int:
        return hash((Scalar, self.name, self.nullable))

Ref

A reference to a named record.

Source code in omnist/schema.py

class Ref:
    """A reference to a named record."""

    __slots__ = ("name",)

    def __init__(self, name: str) -> None:
        self.name = name

    def __repr__(self) -> str:
        return f"ref({self.name})"

    def __eq__(self, other: Any) -> bool:
        return isinstance(other, Ref) and other.name == self.name

    def __hash__(self) -> int:
        return hash((Ref, self.name))

Field

One named, cardinality-bound field slot of a record: label of type, occurring [min, max] times (max=None is unbounded).

Source code in omnist/schema.py

class Field:
    """One named, cardinality-bound field slot of a record: ``label`` of
    ``type``, occurring ``[min, max]`` times (``max=None`` is unbounded)."""

    __slots__ = ("label", "type", "min", "max")

    def __init__(self, label: str, type: Type, min: int = 1,
                 max: Optional[int] = 1) -> None:
        if not isinstance(type, (Ref, Scalar, AnyType)):
            raise SchemaError(
                f"field {label!r} type must be a Ref, Scalar, or t.any, got {type!r}")
        if min < 0 or (max is not None and max < min):
            raise SchemaError(f"field {label!r} has an invalid cardinality [{min},{max}]",
                              code="schema.invalid-cardinality")
        # Note: [0,0] is deliberately still legal to construct directly here
        # (issue #322 only makes it illegal in OSD *text* -- osd.py's
        # _field() -- not as a Python-model invariant; prune()/minimize()
        # construct/consume Field(..., 0, 0) internally as a "dead field"
        # marker, confirmed by tests/test_canonical.py's TestEmptySchemas).
        self.label = label
        self.type = type
        self.min = min
        self.max = max

    def cardinality_str(self) -> str:
        if (self.min, self.max) == (1, 1):
            return "exactly 1"
        if (self.min, self.max) == (0, 1):
            return "0 or 1"
        if self.max is None:
            return f"at least {self.min}"
        return f"between {self.min} and {self.max}"

    def __repr__(self) -> str:
        hi = "" if self.max is None else self.max
        return f"Field({self.label!r}[{self.min},{hi}]: {self.type!r})"

    def __eq__(self, other: Any) -> bool:
        return (isinstance(other, Field) and self.label == other.label
                and self.type == other.type and self.min == other.min
                and self.max == other.max)

ValidationResult

Source code in omnist/schema.py

class ValidationResult:
    def __init__(self) -> None:
        self.errors: List[Error] = []

    @property
    def ok(self) -> bool:
        return not self.errors

    def add(self, path: str, message: str, code: str) -> None:
        self.errors.append(Error(path, message, code))

    def __bool__(self) -> bool:
        return self.ok

    def __str__(self) -> str:
        if self.ok:
            return "valid"
        return "invalid:\n" + "\n".join(f"  at {e.path}: {e.message}" for e in self.errors)

    def __repr__(self) -> str:
        return f"ValidationResult(ok={self.ok}, errors={len(self.errors)})"

Error

Bases: NamedTuple

Source code in omnist/schema.py

class Error(NamedTuple):
    path: str
    message: str
    # stable machine-readable code: validate.unexpected-field, validate.cardinality,
    # validate.type-mismatch, validate.null-not-allowed, validate.shape-mismatch
    code: str

AnyType

The any type: accepts every legal Document value. Singleton; use t.any. Not a Scalar (it has no kind and no nullable flag — null is already included) and not a Ref (it names nothing).

Source code in omnist/schema.py

class AnyType:
    """The `any` type: accepts every legal Document value. Singleton;
    use ``t.any``. Not a Scalar (it has no kind and no nullable flag —
    null is already included) and not a Ref (it names nothing)."""
    __slots__ = ()
    def __eq__(self, other: object) -> bool: return isinstance(other, AnyType)
    def __hash__(self) -> int: return hash(AnyType)
    def __repr__(self) -> str: return "t.any"

AnyFallback dataclass

A single field infer opened as any under allow_any.

location reads RecordName.label; reason says why the field could not be given a single precise type.

Source code in omnist/infer.py

@dataclass(frozen=True)
class AnyFallback:
    """A single field ``infer`` opened as ``any`` under ``allow_any``.

    ``location`` reads ``RecordName.label``; ``reason`` says why the field
    could not be given a single precise type.
    """

    location: str
    reason: str

LintFinding dataclass

One structural diagnostic. code is a stable machine-readable identifier (lint.unsatisfiable-record, lint.unreachable-record, lint.duplicate-record, lint.any-field); severity is warning or info; location is a record name (or Record.label for lint.any-field); message is a human-readable, actionable description.

Source code in omnist/ops/lint.py

@dataclass(frozen=True)
class LintFinding:
    """One structural diagnostic. ``code`` is a stable machine-readable
    identifier (``lint.unsatisfiable-record``, ``lint.unreachable-record``,
    ``lint.duplicate-record``, ``lint.any-field``); ``severity`` is ``warning`` or
    ``info``; ``location`` is a record name (or ``Record.label`` for
    ``lint.any-field``); ``message`` is a human-readable, actionable description."""

    code: str
    severity: str
    location: str
    message: str

WriteReport

Everything a writer adjusted. Truthy when there are no error-severity entries (warnings are fine), so if check_toml(doc): … reads as 'safe'.

Source code in omnist/report.py

class WriteReport:
    """Everything a writer adjusted.  Truthy when there are no error-severity
    entries (warnings are fine), so ``if check_toml(doc): …`` reads as 'safe'."""

    def __init__(self) -> None:
        """Initialize an empty WriteReport."""
        self.adjustments: List[Adjustment] = []

    def add(self, path: str, code: str, message: str, severity: str) -> None:
        """Record an adjustment into this report."""
        self.adjustments.append(Adjustment(path, code, message, severity))

    @property
    def warnings(self) -> List[Adjustment]:
        """All adjustments with ``'warning'`` severity."""
        return [a for a in self.adjustments if a.severity == "warning"]

    @property
    def errors(self) -> List[Adjustment]:
        """All adjustments with ``'error'`` severity."""
        return [a for a in self.adjustments if a.severity == "error"]

    def __bool__(self) -> bool:
        return not self.errors

    def __iter__(self) -> Iterator[Adjustment]:
        return iter(self.adjustments)

    def __len__(self) -> int:
        return len(self.adjustments)

    def __str__(self) -> str:
        if not self.adjustments:
            return "no adjustments"
        return "\n".join(f"{a.severity}: {a.path}: {a.message}" for a in self.adjustments)

errors property

All adjustments with 'error' severity.

warnings property

All adjustments with 'warning' severity.

__init__()

Initialize an empty WriteReport.

Source code in omnist/report.py

def __init__(self) -> None:
    """Initialize an empty WriteReport."""
    self.adjustments: List[Adjustment] = []

add(path, code, message, severity)

Record an adjustment into this report.

Source code in omnist/report.py

def add(self, path: str, code: str, message: str, severity: str) -> None:
    """Record an adjustment into this report."""
    self.adjustments.append(Adjustment(path, code, message, severity))

Adjustment

Bases: NamedTuple

Source code in omnist/report.py

class Adjustment(NamedTuple):
    path: str        # e.g. "$.order.total" — same path style as validation
    code: str        # stable, machine-checkable, e.g. "null.omitted"
    message: str     # human-readable sentence
    severity: str    # "warning" | "error"

Format

Bases: NamedTuple

Source code in omnist/registry.py

class Format(NamedTuple):
    name: str
    read: Callable[[str], Any]                  # text -> node
    write: Callable[..., str]                   # (node, **opts) -> text
    check: Optional[Callable[[Any], Any]] = None  # node -> WriteReport

OmnistError

Bases: Exception

Base class for all omnist errors.

Source code in omnist/errors.py

class OmnistError(Exception):
    """Base class for all omnist errors."""

SchemaError

Bases: OmnistError

The schema text or structure is invalid.

code/path are optional structured attributes -- None unless the raiser passed them, so every existing raise SchemaError("msg") call keeps working unchanged. Where set, code is one of omnist-spec's parse.*/schema.* taxonomy codes (see docs/08-conformance-and-errors.md Sec8.3.1/8.3.3 in the omnist-spec submodule) and path is the OSD text offset or record/field name the problem was found at. Unlike :class:ParseError, a single SchemaError always represents exactly one problem -- OSD parsing stops at the first error, so there is no .errors list to collect (issue #301).

Source code in omnist/errors.py

class SchemaError(OmnistError):
    """The schema text or structure is invalid.

    ``code``/``path`` are optional structured attributes -- ``None`` unless
    the raiser passed them, so every existing ``raise SchemaError("msg")``
    call keeps working unchanged. Where set, ``code`` is one of
    ``omnist-spec``'s ``parse.*``/``schema.*`` taxonomy codes (see
    ``docs/08-conformance-and-errors.md`` Sec8.3.1/8.3.3 in the
    ``omnist-spec`` submodule) and ``path`` is the OSD text offset or
    record/field name the problem was found at. Unlike :class:`ParseError`,
    a single ``SchemaError`` always represents exactly one problem -- OSD
    parsing stops at the first error, so there is no ``.errors`` list to
    collect (issue #301).
    """

    def __init__(self, message: str, *, code: "Optional[str]" = None,
                 path: "Optional[str]" = None) -> None:
        """Initialize SchemaError with a human-readable message and, optionally,
        a structured machine-readable code and the path/position it applies to."""
        super().__init__(message)
        self.code = code
        self.path = path

__init__(message, *, code=None, path=None)

Initialize SchemaError with a human-readable message and, optionally, a structured machine-readable code and the path/position it applies to.

Source code in omnist/errors.py

def __init__(self, message: str, *, code: "Optional[str]" = None,
             path: "Optional[str]" = None) -> None:
    """Initialize SchemaError with a human-readable message and, optionally,
    a structured machine-readable code and the path/position it applies to."""
    super().__init__(message)
    self.code = code
    self.path = path

ParseError

Bases: OmnistError

A document could not be read from its format (outside the supported profile).

Format-syntax failures (invalid JSON/YAML/TOML/XML/OML text) carry code/path (issue #308) -- optional structured attributes, None unless the raiser passed them, so every existing raise ParseError("msg") call keeps working unchanged -- but .errors stays empty, the same way :class:SchemaError distinguishes a single lexical/well-formedness problem from a collected list: a syntax failure stops parsing at the first error, so there is nothing to collect. Schema-conformance failures from :func:~omnist.deserialize.materialize go the other way: they carry the full structured .errors list of every problem found (path, message, machine-readable code), not just the first one, so callers -- an API server turning this into a JSON error response, for instance -- can inspect and report on each one individually instead of parsing str(exc); code/path stay unset for this case, since there's no single position to point at.

Source code in omnist/errors.py

class ParseError(OmnistError):
    """A document could not be read from its format (outside the supported profile).

    Format-syntax failures (invalid JSON/YAML/TOML/XML/OML text) carry
    ``code``/``path`` (issue #308) -- optional structured attributes, ``None``
    unless the raiser passed them, so every existing ``raise
    ParseError("msg")`` call keeps working unchanged -- but ``.errors`` stays
    empty, the same way :class:`SchemaError` distinguishes a single
    lexical/well-formedness problem from a collected list: a syntax failure
    stops parsing at the first error, so there is nothing to collect.
    Schema-conformance failures from :func:`~omnist.deserialize.materialize`
    go the other way: they carry the full structured ``.errors`` list of
    every problem found (path, message, machine-readable code), not just the
    first one, so callers -- an API server turning this into a JSON error
    response, for instance -- can inspect and report on each one
    individually instead of parsing ``str(exc)``; ``code``/``path`` stay
    unset for this case, since there's no single position to point at.
    """

    def __init__(self, message: str, errors: "Optional[List[Error]]" = None, *,
                 code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
        """Initialize ParseError with a human-readable message and either
        structured per-problem issues (materialize) or a structured
        code/path for a single syntax failure -- never both at once."""
        super().__init__(message)
        self.errors: "List[Error]" = errors or []
        self.code = code
        self.path = path

__init__(message, errors=None, *, code=None, path=None)

Initialize ParseError with a human-readable message and either structured per-problem issues (materialize) or a structured code/path for a single syntax failure -- never both at once.

Source code in omnist/errors.py

def __init__(self, message: str, errors: "Optional[List[Error]]" = None, *,
             code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
    """Initialize ParseError with a human-readable message and either
    structured per-problem issues (materialize) or a structured
    code/path for a single syntax failure -- never both at once."""
    super().__init__(message)
    self.errors: "List[Error]" = errors or []
    self.code = code
    self.path = path

WriteError

Bases: OmnistError

A document cannot be represented in the target format.

Raised in strict=True mode for any recorded adjustment, and unconditionally (regardless of strict) when the value has no legal representation at all in the target format -- see docs/08-conformance-and-errors.md Sec8.3.8/8.3.9 in the omnist-spec submodule -- carrying code="write.unsupported-value" and the offending path (issues #323/#324/#325). code/path are optional structured attributes, None unless the raiser passed them, so every existing raise WriteError("msg") call keeps working unchanged. .report holds the full :class:~omnist.report.WriteReport of every adjustment that would have been needed (empty for an unconditional failure raised before any adjustment was recorded), so callers can inspect the structured list, not just the text.

Source code in omnist/errors.py

class WriteError(OmnistError):
    """A document cannot be represented in the target format.

    Raised in ``strict=True`` mode for any recorded adjustment, and
    unconditionally (regardless of ``strict``) when the value has no legal
    representation at all in the target format -- see
    ``docs/08-conformance-and-errors.md`` Sec8.3.8/8.3.9 in the
    ``omnist-spec`` submodule -- carrying ``code="write.unsupported-value"``
    and the offending ``path`` (issues #323/#324/#325). ``code``/``path`` are
    optional structured attributes, ``None`` unless the raiser passed them,
    so every existing ``raise WriteError("msg")`` call keeps working
    unchanged. ``.report`` holds the full
    :class:`~omnist.report.WriteReport` of every adjustment that would have
    been needed (empty for an unconditional failure raised before any
    adjustment was recorded), so callers can inspect the structured list,
    not just the text.
    """

    def __init__(self, message: str, report: "WriteReport | None" = None, *,
                 code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
        """Initialize WriteError with an adjustment report and, optionally,
        a structured machine-readable code and the path it applies to."""
        super().__init__(message)
        self.report = report
        self.code = code
        self.path = path

__init__(message, report=None, *, code=None, path=None)

Initialize WriteError with an adjustment report and, optionally, a structured machine-readable code and the path it applies to.

Source code in omnist/errors.py

def __init__(self, message: str, report: "WriteReport | None" = None, *,
             code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
    """Initialize WriteError with an adjustment report and, optionally,
    a structured machine-readable code and the path it applies to."""
    super().__init__(message)
    self.report = report
    self.code = code
    self.path = path

DocumentError

Bases: OmnistError

A Python value is not a legal Document, or a Document operation is invalid.

Raised by the :class:~omnist.document.Doc API when an import or mutation would produce something outside the Document model — an unsupported Python type, a non-string object key, a cycle — or when an operation doesn't fit the node (e.g. get on a scalar). The message carries the offending path.

code/path are optional structured attributes, None unless the raiser passed them. The reader-side failures that have a document.* code (docs/08-conformance-and-errors.md Sec8.3.2: a safety limit exceeded, an input construct with no label to become an edge) set them; path is then a Document path (E-11), never a text position.

Source code in omnist/errors.py

class DocumentError(OmnistError):
    """A Python value is not a legal Document, or a Document operation is invalid.

    Raised by the :class:`~omnist.document.Doc` API when an import or mutation
    would produce something outside the Document model — an unsupported Python
    type, a non-string object key, a cycle — or when an operation doesn't fit the
    node (e.g. ``get`` on a scalar).  The message carries the offending path.

    ``code``/``path`` are optional structured attributes, ``None`` unless the
    raiser passed them. The reader-side failures that have a ``document.*``
    code (``docs/08-conformance-and-errors.md`` Sec8.3.2: a safety limit
    exceeded, an input construct with no label to become an edge) set them;
    ``path`` is then a Document path (E-11), never a text position.
    """

    def __init__(self, message: str, *, code: "Optional[str]" = None,
                 path: "Optional[str]" = None) -> None:
        """Initialize DocumentError with a human-readable message and,
        optionally, a structured code and the Document path it applies to."""
        super().__init__(message)
        self.code = code
        self.path = path

__init__(message, *, code=None, path=None)

Initialize DocumentError with a human-readable message and, optionally, a structured code and the Document path it applies to.

Source code in omnist/errors.py

def __init__(self, message: str, *, code: "Optional[str]" = None,
             path: "Optional[str]" = None) -> None:
    """Initialize DocumentError with a human-readable message and,
    optionally, a structured code and the Document path it applies to."""
    super().__init__(message)
    self.code = code
    self.path = path

DetachedNode

Bases: DocumentError

A cursor was used after its node was removed from the document.

Holding a :class:~omnist.document.Doc cursor and then removing that node (or a node above it) leaves the cursor pointing at a subtree no longer in the document. Using it raises this instead of silently editing an orphan.

Source code in omnist/errors.py

class DetachedNode(DocumentError):
    """A cursor was used after its node was removed from the document.

    Holding a :class:`~omnist.document.Doc` cursor and then removing that node
    (or a node above it) leaves the cursor pointing at a subtree no longer in the
    document.  Using it raises this instead of silently editing an orphan.
    """

UnsafeXMLWarning

Bases: UserWarning

Unused by read_xml as of the fix for the fail-open XML fallback (see issue #173) — defusedxml is now a hard requirement for XML support, and its absence raises ImportError instead of falling back to the unsafe standard-library parser with a warning. Kept exported for backward compatibility with any code that imports or references it (e.g. an existing warnings.filterwarnings(..., category=omnist.UnsafeXMLWarning) call), but nothing in omnist raises it anymore.

Source code in omnist/errors.py

class UnsafeXMLWarning(UserWarning):
    """Unused by ``read_xml`` as of the fix for the fail-open XML fallback
    (see issue #173) — ``defusedxml`` is now a hard requirement for XML
    support, and its absence raises ``ImportError`` instead of falling back
    to the unsafe standard-library parser with a warning. Kept exported for
    backward compatibility with any code that imports or references it
    (e.g. an existing ``warnings.filterwarnings(..., category=omnist.UnsafeXMLWarning)``
    call), but nothing in omnist raises it anymore.
    """

doc(value)

Build a :class:Doc from a plain Python value.

Source code in omnist/document.py

def doc(value: Any) -> Doc:
    """Build a :class:`Doc` from a plain Python value."""
    return value if isinstance(value, Doc) else Doc.of(value)

record(*fields)

Source code in omnist/schema.py

def record(*fields: Field) -> Record:
    return Record(list(fields))

ref(name)

Source code in omnist/schema.py

def ref(name: str) -> Ref:
    return Ref(name)

field(label, type, min=1, max=1)

Source code in omnist/schema.py

def field(label: str, type: Type, min: int = 1, max: Optional[int] = 1) -> Field:
    return Field(label, type, min, max)

nullable(scalar)

A copy of scalar that also accepts null (the ? form).

Source code in omnist/schema.py

def nullable(scalar: Scalar) -> Scalar:
    """A copy of ``scalar`` that also accepts ``null`` (the ``?`` form)."""
    if isinstance(scalar, AnyType):
        raise SchemaError("any already includes null; 'any?' is redundant",
                          code="schema.nullable-any")
    if isinstance(scalar, Ref):
        raise SchemaError(
            "nullable() cannot be applied to a Ref; use cardinality [0,1] "
            "for an optional record", code="schema.nullable-ref")
    return scalar if scalar.nullable else Scalar(scalar.name, True)

parse_schema(text)

Parse OSD text into a :class:~omnist.schema.Schema.

Source code in omnist/osd.py

def parse_schema(text: str) -> Schema:
    """Parse OSD text into a :class:`~omnist.schema.Schema`."""
    text = strip_bom(text)   # Sec2.5 D-15; a second mark fails as a stray character (D-21)
    return _Parser(_tokenize(text), text).parse()

to_osd(schema, *, indent=4)

Serialize a Schema back to OSD text.

indent=None renders a single-line, machine-oriented form (record defs and the root statement joined by spaces, fields joined by ", ", no trailing comma) instead of the default pretty-printed, indented form -- mirroring write_oml/write_json's own indent=None convention. A non-None int sets the pretty-mode indent width (default 4, matching the prior hardcoded behavior). Both forms round-trip through parse_schema.

A field label is written between double quotes with exactly two escapes (Sec5.9, OSD-15): a backslash is doubled and a double quote gets a backslash in front of it, nothing else -- OSD's unescaping is weak, so those two are all that is needed for every label to read back as itself. A label with a C0 control character (below U+0020) has no OSD spelling at all, so writing it raises :class:~omnist.errors.WriteError with code="write.unsupported-value" and path the Schema path of the record holding the field (OSD-14, E-26), unconditionally.

Source code in omnist/osd.py

def to_osd(schema: Schema, *, indent: Optional[int] = 4) -> str:
    """Serialize a Schema back to OSD text.

    ``indent=None`` renders a single-line, machine-oriented form (record
    defs and the ``root`` statement joined by spaces, fields joined by
    ``", "``, no trailing comma) instead of the default pretty-printed,
    indented form -- mirroring ``write_oml``/``write_json``'s own
    ``indent=None`` convention. A non-``None`` int sets the pretty-mode
    indent width (default 4, matching the prior hardcoded behavior). Both
    forms round-trip through ``parse_schema``.

    A field label is written between double quotes with exactly two
    escapes (Sec5.9, OSD-15): a backslash is doubled and a double quote gets
    a backslash in front of it, nothing else -- OSD's unescaping is weak, so
    those two are all that is needed for every label to read back as itself.
    A label with a C0
    control character (below ``U+0020``) has no OSD spelling at all, so
    writing it raises :class:`~omnist.errors.WriteError` with
    ``code="write.unsupported-value"`` and ``path`` the Schema path of the
    *record* holding the field (OSD-14, E-26), unconditionally.
    """
    parts: List[str] = [_record(name, rec, indent) for name, rec in schema.env.items()]
    parts.append(f"root {schema.root.name}")
    if indent is None:
        return " ".join(parts) + "\n"
    return "\n".join(parts) + "\n"

infer_with_report(samples, root_name='Root', *, allow_any=False)

Source code in omnist/infer.py

def infer_with_report(
    samples: List[Any], root_name: str = "Root", *, allow_any: bool = False,
) -> tuple[Schema, list[AnyFallback]]:
    nodes = []
    for s in samples:
        nodes.append(s._node if isinstance(s, Doc) else build_node(s))
    if not nodes:
        raise SchemaError("cannot infer a schema from zero samples",
                          code="algebra.infer-no-samples", path="$")
    if any(not isinstance(n, list) for n in nodes):
        raise SchemaError("infer expects object (record) samples at the root",
                          code="algebra.infer-scalar-root", path="$")
    env: Dict[str, Any] = {}
    used: set[str] = set()
    fallbacks: list[AnyFallback] = []
    _infer_record(nodes, root_name, env, used, allow_any, fallbacks, 0)
    return Schema(Ref(root_name), env), fallbacks

materialize(node, schema)

A copy of node with leaf values upgraded to match schema, guaranteed to conform to it -- raises :class:~omnist.errors.ParseError (with every problem found, not just the first, in both the message and the structured .errors list) if it can't be made to.

Source code in omnist/deserialize.py

def materialize(node: Any, schema: Schema) -> Any:
    """A copy of ``node`` with leaf values upgraded to match ``schema``,
    guaranteed to conform to it -- raises :class:`~omnist.errors.ParseError`
    (with every problem found, not just the first, in both the message and
    the structured ``.errors`` list) if it can't be made to."""
    res = ValidationResult()
    out = _materialize_type(node, schema, schema.root, "$", res)
    if not res.ok:
        raise ParseError(str(res), errors=res.errors)
    return out

read_json(text, *, schema=None)

Source code in omnist/formats.py

def read_json(text: str, *, schema: Optional["Schema"] = None) -> Any:
    text = strip_bom(text, reject_second=True)   # Sec2.5 D-15/D-21
    _check_json_text_depth(text)
    try:
        node = build_node(_json.loads(text))
    except _json.JSONDecodeError as exc:
        raise ParseError(f"invalid JSON: {exc}", code="parse.codec-syntax",
                         path=f"{exc.lineno}:{exc.colno}") from exc
    except ValueError as exc:
        # json.loads converts integer literals to `int` while parsing, so an
        # over-digit-limit literal trips CPython's int-string-conversion
        # guard here -- before build_node ever sees a value -- as a bare
        # ValueError, not a JSONDecodeError.  That is the int-digits limit
        # (D-13), reported like build_node's own.
        raise _int_digits_limit(f"invalid JSON: {exc}") from exc
    return _materialize(node, schema)

write_json(node, *, indent=None, strict=False, report=None)

Source code in omnist/formats.py

def write_json(node: Any, *, indent: Optional[int] = None, strict: bool = False,
               report: Optional[WriteReport] = None) -> str:
    rep = _scan_json(node)
    text = _json.dumps(_grouped(node), indent=indent, ensure_ascii=False, default=_iso)
    return finish_write(text, rep, strict=strict, report=report)

read_yaml(text, *, schema=None)

Source code in omnist/formats.py

def read_yaml(text: str, *, schema: Optional["Schema"] = None) -> Any:
    text = strip_bom(text, reject_second=True)   # Sec2.5 D-15/D-21
    yaml = _need("yaml", "pip install pyyaml")
    try:
        node = build_node(yaml.load(text, Loader=_yaml_loader(yaml)))
    except yaml.YAMLError as exc:
        raise ParseError(f"invalid YAML: {exc}", code="parse.codec-syntax",
                         path=_yaml_position(exc, text)) from exc
    except ValueError as exc:
        # Same int-string-conversion guard as read_json (see its comment):
        # PyYAML converts an integer scalar to `int` while loading.
        raise _int_digits_limit(f"invalid YAML: {exc}") from exc
    except RecursionError as exc:
        # #307: unlike JSON's bracket grammar, YAML nesting (indentation,
        # flow collections, anchors) isn't cheap to bound from raw text
        # without reimplementing the grammar -- this is a safety net that
        # converts an uncaught crash into the same clean error a depth
        # violation always raises, rather than precise prevention.
        raise ParseError(f"nesting exceeds the maximum depth ({_MAX_DEPTH})",
                         code="document.limit.depth", path="$") from exc
    return _materialize(node, schema)

write_yaml(node, *, strict=False, report=None)

Source code in omnist/formats.py

def write_yaml(node: Any, *, strict: bool = False,
               report: Optional[WriteReport] = None) -> str:
    yaml = _need("yaml", "pip install pyyaml")
    rep = check_yaml(node)
    prepared = _prepare_yaml(node)
    dumper = _yaml_dumper(yaml)
    text = yaml.dump(_grouped(prepared), Dumper=dumper, sort_keys=False,
                     allow_unicode=True, default_flow_style=False)
    return finish_write(text, rep, strict=strict, report=report)

read_toml(text, *, schema=None)

Source code in omnist/formats.py

def read_toml(text: str, *, schema: Optional["Schema"] = None) -> Any:
    text = strip_bom(text, reject_second=True)   # Sec2.5 D-15/D-21
    import tomllib
    try:
        node = build_node(tomllib.loads(text))
    except tomllib.TOMLDecodeError as exc:
        raise ParseError(f"invalid TOML: {exc}", code="parse.codec-syntax",
                         path=_toml_position(exc, text)) from exc
    except ValueError as exc:
        # Same int-string-conversion guard as read_json (see its comment):
        # tomllib converts an integer literal to `int` while loading.
        raise _int_digits_limit(f"invalid TOML: {exc}") from exc
    except RecursionError as exc:
        # #307: same safety net as read_yaml -- TOML nesting (inline
        # tables/arrays, dotted keys) isn't cheap to bound from raw text
        # without reimplementing the grammar.
        raise ParseError(f"nesting exceeds the maximum depth ({_MAX_DEPTH})",
                         code="document.limit.depth", path="$") from exc
    return _materialize(node, schema)

write_toml(node, *, strict=False, report=None)

Source code in omnist/formats.py

def write_toml(node: Any, *, strict: bool = False,
               report: Optional[WriteReport] = None) -> str:
    tomli_w = _need("tomli_w", "pip install tomli_w")
    rep = WriteReport()
    stripped = _strip_nulls(node, "$", rep)        # TOML has no null
    _check_interleaving(node, rep)
    grouped = _grouped(stripped)
    if not isinstance(grouped, dict):
        raise WriteError("TOML needs a top-level table (the root must be an object)")
    text = tomli_w.dumps(grouped)
    return finish_write(text, rep, strict=strict, report=report)

read_xml(text, *, schema=None, report=None)

Source code in omnist/formats.py

def read_xml(text: str, *, schema: Optional["Schema"] = None,
            report: Optional[WriteReport] = None) -> Any:
    text = strip_bom(text, reject_second=True)   # Sec2.5 D-15/D-21
    root = _xml_fromstring(text)
    root_local = _local(root.tag)
    # `epath` is the proper full dotted Document path (Sec8.4 convention),
    # used only for format.attribute-dropped/format.namespace-dropped --
    # kept separate from `path` below, which is the pre-existing (shallower)
    # scheme mixed-content errors already use and report on.
    root_epath = f"$.{root_local}"
    _check_xml_drops(root, root_epath, report)
    node = [(root_local, _xml_to_node(root, "$", 0, [0], report, root_epath))]
    if schema is not None:
        # #288: recover boolean/integer/number from XML's untyped text
        # before the shared materialize() sees it. This is XML-specific,
        # not a materialize() capability -- materialize() itself must keep
        # rejecting a string that merely looks numeric (a string is a
        # deliberate choice in JSON/YAML/TOML/OML, never an untyped
        # placeholder the way it always is in XML).
        node = _xml_pretype(node, schema, schema.root)
    return _materialize(node, schema)

write_xml(node, *, strict=False, report=None)

Source code in omnist/formats.py

def write_xml(node: Any, *, strict: bool = False,
              report: Optional[WriteReport] = None) -> str:
    if not (isinstance(node, list) and len(node) == 1):
        raise WriteError(
            "XML needs exactly one document element; the root node must have a "
            "single top-level edge (a single-rooted Document)",
            code="format.multiple-roots", path="$")
    rep = check_xml(node)
    import xml.etree.ElementTree as ET
    (tag, content), = node
    el = ET.Element(tag)
    _node_to_xml(content, el)
    _indent(el)
    text = ET.tostring(el, encoding="unicode")
    # Issue #326: a numeric character reference is exempt from XML's
    # mandated line-ending normalization on parse, so escaping a literal CR
    # (and CRLF) this way is genuinely lossless -- unlike every other
    # `\r`-in-text-content byte, a real `
` never occurs in
    # ET.tostring's own output (its own formatting/indentation only ever
    # inserts `\n`), so a plain string replace on the serialized text can't
    # collide with anything else in it.
    text = text.replace("\r\n", "
\n").replace("\r", "
")
    return finish_write(text, rep, strict=strict, report=report)

read_oml(text, *, schema=None)

Parse OML source into a canonical Document node (edge-list or leaf).

Source code in omnist/oml.py

def read_oml(text: str, *, schema: Optional[Any] = None) -> Any:
    """Parse OML source into a canonical Document node (edge-list or leaf)."""
    scanner = _Scanner(strip_bom(text))   # Sec2.5 D-15
    node = _Parser(scanner).parse_document()
    if schema is None:
        return node
    from .deserialize import materialize
    return materialize(node, schema)

write_oml(node, *, indent=2, arrays=False)

Render a canonical Document node as OML source.

OML is lossless for every Document: there is never an adjustment to report (unlike JSON/YAML/TOML/XML), so there is no check_oml/ strict=/report= machinery — the write always succeeds exactly.

indent=None renders a single-line, machine-oriented form (edges joined by "; ", no newlines/padding) instead of the default pretty-printed, indented form -- mirroring write_json's own indent=None convention. Both forms round-trip through read_oml.

arrays=True (issue #218) collapses any maximal run of >= 2 consecutive same-label edges into label: [v1, v2, ...] array syntax -- a run of length 1 still writes as a plain scalar edge, and a run is never merged across an edge with a different label in between, so this never reorders anything: read_oml(write_oml(node, arrays=True)) == node holds unconditionally. Default False produces output byte-identical to arrays not existing at all.

Source code in omnist/oml.py

def write_oml(node: Any, *, indent: Optional[int] = 2, arrays: bool = False) -> str:
    """Render a canonical Document node as OML source.

    OML is lossless for every Document: there is never an adjustment to
    report (unlike JSON/YAML/TOML/XML), so there is no ``check_oml``/
    ``strict=``/``report=`` machinery — the write always succeeds exactly.

    ``indent=None`` renders a single-line, machine-oriented form (edges
    joined by ``"; "``, no newlines/padding) instead of the default
    pretty-printed, indented form -- mirroring ``write_json``'s own
    ``indent=None`` convention. Both forms round-trip through ``read_oml``.

    ``arrays=True`` (issue #218) collapses any maximal run of >= 2
    consecutive same-label edges into ``label: [v1, v2, ...]`` array
    syntax -- a run of length 1 still writes as a plain scalar edge, and a
    run is never merged across an edge with a different label in between,
    so this never reorders anything: ``read_oml(write_oml(node,
    arrays=True)) == node`` holds unconditionally. Default ``False``
    produces output byte-identical to ``arrays`` not existing at all.
    """
    if not isinstance(node, list):
        return _write_scalar(node)
    if indent is None:
        return _write_edges_compact(node, arrays, 0)
    return _write_edges(node, 0, indent, arrays, 0)

check_json(node)

Report what writing JSON would adjust, without producing output.

Source code in omnist/formats.py

def check_json(node: Any) -> WriteReport:
    """Report what writing JSON would adjust, without producing output."""
    return _scan_json(node)

check_yaml(node)

Source code in omnist/formats.py

def check_yaml(node: Any) -> WriteReport:
    rep = WriteReport()
    for path, v in _leaves(node):
        if isinstance(v, _dt.time):       # YAML carries date/datetime natively, not time
            rep.add(path, "format.temporal-stringified",
                    "time-of-day written as a string (YAML has no standalone time)",
                    "warning")
    _scan_yaml_labels(node, "$", rep)
    _check_interleaving(node, rep)
    return rep

check_toml(node)

Source code in omnist/formats.py

def check_toml(node: Any) -> WriteReport:
    rep = WriteReport()
    _strip_nulls(node, "$", rep)
    _check_interleaving(node, rep)
    return rep

check_xml(node)

Source code in omnist/formats.py

def check_xml(node: Any) -> WriteReport:
    rep = WriteReport()
    _scan_xml(node, "$", rep)
    return rep

check_oml(node)

OML can hold every Document losslessly; always an empty report.

Source code in omnist/oml.py

def check_oml(node: Any) -> "WriteReport":
    """OML can hold every Document losslessly; always an empty report."""
    from .report import WriteReport
    return WriteReport()

finish_write(text, rep, *, strict=False, report=None)

Apply the standard strict / report handling to a writer's result.

If report is given, rep's adjustments are copied into it. If strict and rep has any adjustments, raises WriteError carrying rep. Otherwise returns text.

Source code in omnist/report.py

def finish_write(text: str, rep: WriteReport, *, strict: bool = False,
                 report: Optional[WriteReport] = None) -> str:
    """Apply the standard ``strict`` / ``report`` handling to a writer's result.

    If ``report`` is given, ``rep``'s adjustments are copied into it.  If
    ``strict`` and ``rep`` has any adjustments, raises ``WriteError`` carrying
    ``rep``.  Otherwise returns ``text``.
    """
    if report is not None:
        report.adjustments.extend(rep.adjustments)
    if strict and rep.adjustments:
        raise WriteError(str(rep), report=rep)
    return text

register_format(fmt)

Register (or replace) a format plugin.

Source code in omnist/registry.py

def register_format(fmt: Format) -> None:
    """Register (or replace) a format plugin."""
    with _LOCK:
        _REGISTRY[fmt.name] = fmt

get_format(name)

The registered :class:Format for name (raises if unknown).

Source code in omnist/registry.py

def get_format(name: str) -> Format:
    """The registered :class:`Format` for ``name`` (raises if unknown)."""
    with _LOCK:
        try:
            return _REGISTRY[name]
        except KeyError:
            known = ", ".join(sorted(_REGISTRY)) or "(none)"
            raise OmnistError(f"unknown format {name!r}; registered: {known}") from None

---

Document Model (omnist.document)

The Document model — a canonical tree of ordered, labeled edges.

A Document model node is either

• a leaf holding a scalar value (str/int/float/bool/ datetime values, or None), or
• an internal node holding an ordered list of edges, each a (label, child) pair. Labels may repeat — "many members" is the label member appearing several times, not a field pointing to an array.

The canonical Python form of a node is therefore::

scalar                                   # a leaf
[(label, node), (label, node), ...]      # an internal node (ordered)

This single shape represents every supported format canonically, including XML's interleaved repeated elements, which a dict-with-array-values cannot. Doc is a thin, guarded wrapper around a node, with navigation helpers. Order is preserved (it is data); schema validation ignores it. See docs/design/model.md.

Doc

A guarded handle on a Document node (a leaf value or an edge list).

Source code in omnist/document.py

class Doc:
    """A guarded handle on a Document node (a leaf value or an edge list)."""

    __slots__ = ("_node", "path", "depth")

    def __init__(self, node: Any, path: str = "$", depth: int = 0) -> None:
        """Initialize a Doc cursor pointing to ``node`` at ``path``."""
        self._node = node
        self.path = path
        self.depth = depth

    # -- construction ---------------------------------------------------
    @classmethod
    def of(cls, value: Any) -> "Doc":
        """Construct a Doc from a plain Python value (JSON-shaped mapping, sequence, or scalar)."""
        return cls(build_node(value))

    @classmethod
    def from_format(cls, name: str, text: str) -> "Doc":
        """Parse source text using the registered format named ``name`` into a Doc."""
        from .registry import get_format
        return cls(get_format(name).read(text))

    @classmethod
    def from_json(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse JSON text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_json
        return cls(read_json(text, schema=schema))

    @classmethod
    def from_yaml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse YAML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_yaml
        return cls(read_yaml(text, schema=schema))

    @classmethod
    def from_toml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse TOML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_toml
        return cls(read_toml(text, schema=schema))

    @classmethod
    def from_xml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse XML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .formats import read_xml
        return cls(read_xml(text, schema=schema))

    @classmethod
    def from_oml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
        """Parse OML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
        from .oml import read_oml
        return cls(read_oml(text, schema=schema))

    # -- shape ----------------------------------------------------------
    @property
    def is_leaf(self) -> bool:
        """True if this cursor points to a scalar leaf value (spec §2.1)."""
        return not isinstance(self._node, list)

    @property
    def value(self) -> Any:
        """The scalar value at this leaf node (spec §2.2).

        Raises :class:`~omnist.errors.DocumentError` if this is an internal node.
        """
        if isinstance(self._node, list):
            raise DocumentError(f"{self.path}: not a leaf; use edges()")
        return self._node

    def edges(self) -> List[Tuple[str, "Doc"]]:
        """Return the ordered list of ``(label, Doc)`` child edges (spec §2.1).

        Raises :class:`~omnist.errors.DocumentError` if this is a leaf node.
        """
        if not isinstance(self._node, list):
            raise DocumentError(f"{self.path}: a leaf has no edges")
        out: List[Tuple[str, "Doc"]] = []
        counts: dict[str, int] = {}
        for label, child in self._node:
            i = counts.get(label, 0)
            counts[label] = i + 1
            cp = f"{self.path}.{label}" if i == 0 else f"{self.path}.{label}[{i}]"
            out.append((label, Doc(child, cp, self.depth + 1)))
        return out

    def labels(self) -> List[str]:
        """Return the deduplicated list of child edge labels in first-occurrence order."""
        seen: set[str] = set()
        out: List[str] = []
        for label, _ in self._iter():
            if label not in seen:
                seen.add(label)
                out.append(label)
        return out

    def get(self, label: str) -> List["Doc"]:
        """Return all child Doc cursors matching ``label``."""
        return [c for lbl, c in self.edges() if lbl == label]

    def get_one(self, label: str) -> "Doc":
        """Return the single child Doc cursor matching ``label``.

        Raises :class:`~omnist.errors.DocumentError` if there are not exactly 1 matching edge.
        """
        cs = self.get(label)
        if len(cs) != 1:
            raise DocumentError(
                f"{self.path}: expected exactly one {label!r}, found {len(cs)}")
        return cs[0]

    def count(self, label: str) -> int:
        """Return the number of child edges matching ``label``."""
        return sum(1 for lbl, _ in self._iter() if lbl == label)

    def _iter(self) -> Iterator[Tuple[str, Any]]:
        if isinstance(self._node, list):
            yield from self._node

    def child(self, label: str) -> "Doc":
        """A cursor to the single child under ``label`` (editable if internal)."""
        return self.get_one(label)

    # -- editing (mutates the underlying edge list) ---------------------
    def add(self, label: str, value: Any) -> "Doc":
        """Append an edge ``(label, value)``.  A repeated label is how an array
        grows.  Returns ``self`` for chaining."""
        self._require_internal("add")
        self._node.append(
            (label, build_node(value, f"{self.path}.{label}", self.depth + 1)))
        return self

    def remove(self, label: str) -> "Doc":
        """Remove every edge under ``label``."""
        self._require_internal("remove")
        self._node[:] = [(lbl, c) for lbl, c in self._node if lbl != label]
        return self

    def set(self, label: str, value: Any) -> "Doc":
        """Replace all edges under ``label`` with a single new edge (positioned
        at the first old occurrence); ``set`` = ``remove`` + ``add``."""
        self._require_internal("set")
        new = build_node(value, f"{self.path}.{label}", self.depth + 1)
        first = None
        kept: List[Edge] = []
        for lbl, child in self._node:
            if lbl == label:
                if first is None:
                    first = len(kept)
                    kept.append((label, new))
                # later duplicates are dropped
            else:
                kept.append((lbl, child))
        if first is None:
            kept.append((label, new))
        self._node[:] = kept
        return self

    def _require_internal(self, op: str) -> None:
        if not isinstance(self._node, list):
            raise DocumentError(f"{self.path}: cannot {op} on a leaf")

    # -- export ---------------------------------------------------------
    def to_data(self) -> Any:
        """Return a deep copy of the raw underlying node representation (spec §2.1)."""
        return _copy(self._node)

    def to_grouped(self) -> Any:
        """A JSON-shaped projection: same-label edges grouped into a list.

        A label seen once stays a single value; a label seen more than once
        becomes a list (the schema-less fallback of the count-1 rule, see
        ``docs/design/model.md`` §10)."""
        return _grouped(self._node)

    def to_json(self, **o: Any) -> str:
        """Serialize this Document to JSON text (spec §4)."""
        from .formats import write_json
        return write_json(self._node, **o)

    def to_yaml(self, **o: Any) -> str:
        """Serialize this Document to YAML text (spec §4)."""
        from .formats import write_yaml
        return write_yaml(self._node, **o)

    def to_toml(self, **o: Any) -> str:
        """Serialize this Document to TOML text (spec §4)."""
        from .formats import write_toml
        return write_toml(self._node, **o)

    def to_xml(self, **o: Any) -> str:
        """Serialize this Document to XML text (spec §4)."""
        from .formats import write_xml
        return write_xml(self._node, **o)

    def to_oml(self, **o: Any) -> str:
        """Serialize this Document to OML text (spec §4)."""
        from .oml import write_oml
        return write_oml(self._node, **o)

    def to_format(self, name: str, **o: Any) -> str:
        """Serialize this Document to the registered format named ``name``."""
        from .registry import get_format
        return get_format(name).write(self._node, **o)

    def check_json(self) -> "WriteReport":
        """Simulate writing to JSON and return the adjustment report (spec §4)."""
        from .formats import check_json
        return check_json(self._node)

    def check_yaml(self) -> "WriteReport":
        """Simulate writing to YAML and return the adjustment report (spec §4)."""
        from .formats import check_yaml
        return check_yaml(self._node)

    def check_toml(self) -> "WriteReport":
        """Simulate writing to TOML and return the adjustment report (spec §4)."""
        from .formats import check_toml
        return check_toml(self._node)

    def check_xml(self) -> "WriteReport":
        """Simulate writing to XML and return the adjustment report (spec §4)."""
        from .formats import check_xml
        return check_xml(self._node)

    def check_oml(self) -> "WriteReport":
        """Simulate writing to OML and return the adjustment report (spec §4)."""
        from .oml import check_oml
        return check_oml(self._node)

    def check_format(self, name: str) -> "WriteReport":
        """Simulate writing to format ``name`` and return the adjustment
        report, without producing output. Requires the registered
        :class:`~omnist.registry.Format` to provide a ``check``
        callable (the four built-ins do; a custom plugin may not)."""
        from .registry import get_format
        fmt = get_format(name)
        if fmt.check is None:
            raise DocumentError(
                f"format {name!r} has no check() -- cannot simulate a write")
        return fmt.check(self._node)  # type: ignore[no-any-return]

    def validate(self, schema: "Schema") -> "ValidationResult":
        """Validate this Document against ``schema`` (spec §5)."""
        return schema.validate(self)

    # -- dunders --------------------------------------------------------
    def __eq__(self, other: Any) -> bool:
        if isinstance(other, Doc):
            return self._node == other._node  # type: ignore[no-any-return]
        try:
            return self._node == build_node(other)  # type: ignore[no-any-return]
        except DocumentError:
            return NotImplemented

    def __repr__(self) -> str:
        return f"Doc({'leaf' if self.is_leaf else 'node'}: {self._node!r})"

is_leaf property

True if this cursor points to a scalar leaf value (spec §2.1).

value property

The scalar value at this leaf node (spec §2.2).

Raises :class:~omnist.errors.DocumentError if this is an internal node.

__init__(node, path='$', depth=0)

Initialize a Doc cursor pointing to node at path.

Source code in omnist/document.py

def __init__(self, node: Any, path: str = "$", depth: int = 0) -> None:
    """Initialize a Doc cursor pointing to ``node`` at ``path``."""
    self._node = node
    self.path = path
    self.depth = depth

add(label, value)

Append an edge (label, value). A repeated label is how an array grows. Returns self for chaining.

Source code in omnist/document.py

def add(self, label: str, value: Any) -> "Doc":
    """Append an edge ``(label, value)``.  A repeated label is how an array
    grows.  Returns ``self`` for chaining."""
    self._require_internal("add")
    self._node.append(
        (label, build_node(value, f"{self.path}.{label}", self.depth + 1)))
    return self

check_format(name)

Simulate writing to format name and return the adjustment report, without producing output. Requires the registered :class:~omnist.registry.Format to provide a check callable (the four built-ins do; a custom plugin may not).

Source code in omnist/document.py

def check_format(self, name: str) -> "WriteReport":
    """Simulate writing to format ``name`` and return the adjustment
    report, without producing output. Requires the registered
    :class:`~omnist.registry.Format` to provide a ``check``
    callable (the four built-ins do; a custom plugin may not)."""
    from .registry import get_format
    fmt = get_format(name)
    if fmt.check is None:
        raise DocumentError(
            f"format {name!r} has no check() -- cannot simulate a write")
    return fmt.check(self._node)  # type: ignore[no-any-return]

check_json()

Simulate writing to JSON and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_json(self) -> "WriteReport":
    """Simulate writing to JSON and return the adjustment report (spec §4)."""
    from .formats import check_json
    return check_json(self._node)

check_oml()

Simulate writing to OML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_oml(self) -> "WriteReport":
    """Simulate writing to OML and return the adjustment report (spec §4)."""
    from .oml import check_oml
    return check_oml(self._node)

check_toml()

Simulate writing to TOML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_toml(self) -> "WriteReport":
    """Simulate writing to TOML and return the adjustment report (spec §4)."""
    from .formats import check_toml
    return check_toml(self._node)

check_xml()

Simulate writing to XML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_xml(self) -> "WriteReport":
    """Simulate writing to XML and return the adjustment report (spec §4)."""
    from .formats import check_xml
    return check_xml(self._node)

check_yaml()

Simulate writing to YAML and return the adjustment report (spec §4).

Source code in omnist/document.py

def check_yaml(self) -> "WriteReport":
    """Simulate writing to YAML and return the adjustment report (spec §4)."""
    from .formats import check_yaml
    return check_yaml(self._node)

child(label)

A cursor to the single child under label (editable if internal).

Source code in omnist/document.py

def child(self, label: str) -> "Doc":
    """A cursor to the single child under ``label`` (editable if internal)."""
    return self.get_one(label)

count(label)

Return the number of child edges matching label.

Source code in omnist/document.py

def count(self, label: str) -> int:
    """Return the number of child edges matching ``label``."""
    return sum(1 for lbl, _ in self._iter() if lbl == label)

edges()

Return the ordered list of (label, Doc) child edges (spec §2.1).

Raises :class:~omnist.errors.DocumentError if this is a leaf node.

Source code in omnist/document.py

def edges(self) -> List[Tuple[str, "Doc"]]:
    """Return the ordered list of ``(label, Doc)`` child edges (spec §2.1).

    Raises :class:`~omnist.errors.DocumentError` if this is a leaf node.
    """
    if not isinstance(self._node, list):
        raise DocumentError(f"{self.path}: a leaf has no edges")
    out: List[Tuple[str, "Doc"]] = []
    counts: dict[str, int] = {}
    for label, child in self._node:
        i = counts.get(label, 0)
        counts[label] = i + 1
        cp = f"{self.path}.{label}" if i == 0 else f"{self.path}.{label}[{i}]"
        out.append((label, Doc(child, cp, self.depth + 1)))
    return out

from_format(name, text) classmethod

Parse source text using the registered format named name into a Doc.

Source code in omnist/document.py

@classmethod
def from_format(cls, name: str, text: str) -> "Doc":
    """Parse source text using the registered format named ``name`` into a Doc."""
    from .registry import get_format
    return cls(get_format(name).read(text))

from_json(text, *, schema=None) classmethod

Parse JSON text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_json(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse JSON text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_json
    return cls(read_json(text, schema=schema))

from_oml(text, *, schema=None) classmethod

Parse OML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_oml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse OML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .oml import read_oml
    return cls(read_oml(text, schema=schema))

from_toml(text, *, schema=None) classmethod

Parse TOML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_toml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse TOML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_toml
    return cls(read_toml(text, schema=schema))

from_xml(text, *, schema=None) classmethod

Parse XML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_xml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse XML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_xml
    return cls(read_xml(text, schema=schema))

from_yaml(text, *, schema=None) classmethod

Parse YAML text into a Doc, optionally upgrading leaves against schema (spec §4).

Source code in omnist/document.py

@classmethod
def from_yaml(cls, text: str, *, schema: Optional["Schema"] = None) -> "Doc":
    """Parse YAML text into a Doc, optionally upgrading leaves against ``schema`` (spec §4)."""
    from .formats import read_yaml
    return cls(read_yaml(text, schema=schema))

get(label)

Return all child Doc cursors matching label.

Source code in omnist/document.py

def get(self, label: str) -> List["Doc"]:
    """Return all child Doc cursors matching ``label``."""
    return [c for lbl, c in self.edges() if lbl == label]

get_one(label)

Return the single child Doc cursor matching label.

Raises :class:~omnist.errors.DocumentError if there are not exactly 1 matching edge.

Source code in omnist/document.py

def get_one(self, label: str) -> "Doc":
    """Return the single child Doc cursor matching ``label``.

    Raises :class:`~omnist.errors.DocumentError` if there are not exactly 1 matching edge.
    """
    cs = self.get(label)
    if len(cs) != 1:
        raise DocumentError(
            f"{self.path}: expected exactly one {label!r}, found {len(cs)}")
    return cs[0]

labels()

Return the deduplicated list of child edge labels in first-occurrence order.

Source code in omnist/document.py

def labels(self) -> List[str]:
    """Return the deduplicated list of child edge labels in first-occurrence order."""
    seen: set[str] = set()
    out: List[str] = []
    for label, _ in self._iter():
        if label not in seen:
            seen.add(label)
            out.append(label)
    return out

of(value) classmethod

Construct a Doc from a plain Python value (JSON-shaped mapping, sequence, or scalar).

Source code in omnist/document.py

@classmethod
def of(cls, value: Any) -> "Doc":
    """Construct a Doc from a plain Python value (JSON-shaped mapping, sequence, or scalar)."""
    return cls(build_node(value))

remove(label)

Remove every edge under label.

Source code in omnist/document.py

def remove(self, label: str) -> "Doc":
    """Remove every edge under ``label``."""
    self._require_internal("remove")
    self._node[:] = [(lbl, c) for lbl, c in self._node if lbl != label]
    return self

set(label, value)

Replace all edges under label with a single new edge (positioned at the first old occurrence); set = remove + add.

Source code in omnist/document.py

def set(self, label: str, value: Any) -> "Doc":
    """Replace all edges under ``label`` with a single new edge (positioned
    at the first old occurrence); ``set`` = ``remove`` + ``add``."""
    self._require_internal("set")
    new = build_node(value, f"{self.path}.{label}", self.depth + 1)
    first = None
    kept: List[Edge] = []
    for lbl, child in self._node:
        if lbl == label:
            if first is None:
                first = len(kept)
                kept.append((label, new))
            # later duplicates are dropped
        else:
            kept.append((lbl, child))
    if first is None:
        kept.append((label, new))
    self._node[:] = kept
    return self

to_data()

Return a deep copy of the raw underlying node representation (spec §2.1).

Source code in omnist/document.py

def to_data(self) -> Any:
    """Return a deep copy of the raw underlying node representation (spec §2.1)."""
    return _copy(self._node)

to_format(name, **o)

Serialize this Document to the registered format named name.

Source code in omnist/document.py

def to_format(self, name: str, **o: Any) -> str:
    """Serialize this Document to the registered format named ``name``."""
    from .registry import get_format
    return get_format(name).write(self._node, **o)

to_grouped()

A JSON-shaped projection: same-label edges grouped into a list.

A label seen once stays a single value; a label seen more than once becomes a list (the schema-less fallback of the count-1 rule, see docs/design/model.md §10).

Source code in omnist/document.py

def to_grouped(self) -> Any:
    """A JSON-shaped projection: same-label edges grouped into a list.

    A label seen once stays a single value; a label seen more than once
    becomes a list (the schema-less fallback of the count-1 rule, see
    ``docs/design/model.md`` §10)."""
    return _grouped(self._node)

to_json(**o)

Serialize this Document to JSON text (spec §4).

Source code in omnist/document.py

def to_json(self, **o: Any) -> str:
    """Serialize this Document to JSON text (spec §4)."""
    from .formats import write_json
    return write_json(self._node, **o)

to_oml(**o)

Serialize this Document to OML text (spec §4).

Source code in omnist/document.py

def to_oml(self, **o: Any) -> str:
    """Serialize this Document to OML text (spec §4)."""
    from .oml import write_oml
    return write_oml(self._node, **o)

to_toml(**o)

Serialize this Document to TOML text (spec §4).

Source code in omnist/document.py

def to_toml(self, **o: Any) -> str:
    """Serialize this Document to TOML text (spec §4)."""
    from .formats import write_toml
    return write_toml(self._node, **o)

to_xml(**o)

Serialize this Document to XML text (spec §4).

Source code in omnist/document.py

def to_xml(self, **o: Any) -> str:
    """Serialize this Document to XML text (spec §4)."""
    from .formats import write_xml
    return write_xml(self._node, **o)

to_yaml(**o)

Serialize this Document to YAML text (spec §4).

Source code in omnist/document.py

def to_yaml(self, **o: Any) -> str:
    """Serialize this Document to YAML text (spec §4)."""
    from .formats import write_yaml
    return write_yaml(self._node, **o)

validate(schema)

Validate this Document against schema (spec §5).

Source code in omnist/document.py

def validate(self, schema: "Schema") -> "ValidationResult":
    """Validate this Document against ``schema`` (spec §5)."""
    return schema.validate(self)

build_node(value, path='$', depth=0, seen=None, budget=None)

Turn a plain Python value into a canonical node.

A dict becomes an ordered edge list; a key whose value is a list expands into one edge per item (the same label repeated). A scalar becomes a leaf. A bare list (a top-level array, or a list nested directly inside a list) has no labeled-edge form and raises DocumentError.

budget is a shared running count of every value materialized across the whole call tree (unlike seen, which is scoped per ancestor path). seen alone only catches a literal cycle -- a value that is its own ancestor -- not a DAG node reached via two different, non-ancestor paths (e.g. two YAML aliases pointing at the same anchor). Such a shared node gets walked in full at every occurrence, so a chain of n doubly-referenced anchors materializes O(2**n) nodes from an O(n)-sized source text ("billion laughs"). budget bounds that blowup directly, independent of _MAX_DEPTH (which only bounds nesting depth, not breadth/repetition) and independent of any single caller's input format.

Source code in omnist/document.py

def build_node(value: Any, path: str = "$", depth: int = 0,
               seen: Optional[frozenset[int]] = None,
               budget: Optional[List[int]] = None) -> Any:
    """Turn a plain Python value into a canonical node.

    A ``dict`` becomes an ordered edge list; a key whose value is a list expands
    into one edge **per item** (the same label repeated).  A scalar becomes a
    leaf.  A *bare* list (a top-level array, or a list nested directly inside a
    list) has no labeled-edge form and raises ``DocumentError``.

    ``budget`` is a shared running count of every value materialized across
    the *whole* call tree (unlike ``seen``, which is scoped per ancestor
    path). ``seen`` alone only catches a literal cycle -- a value that is its
    own ancestor -- not a DAG node reached via two different, non-ancestor
    paths (e.g. two YAML aliases pointing at the same anchor). Such a shared
    node gets walked in full at *every* occurrence, so a chain of ``n``
    doubly-referenced anchors materializes ``O(2**n)`` nodes from an
    ``O(n)``-sized source text ("billion laughs"). ``budget`` bounds that
    blowup directly, independent of ``_MAX_DEPTH`` (which only bounds
    nesting depth, not breadth/repetition) and independent of any single
    caller's input format.
    """
    if budget is None:
        budget = [0]
    if depth > _MAX_DEPTH:
        raise DocumentError(f"{path}: nesting exceeds the maximum depth ({_MAX_DEPTH})",
                            code="document.limit.depth", path="$")
    if isinstance(value, dict):
        # #309: only an actual container -- omnist-spec Sec2.2's formal
        # `node = [edge, edge, ...]` -- counts against the node budget. A
        # scalar leaf is a `value`, not a `node`, per that same grammar;
        # the vendored test-suite's node-count-at-declared-limit-succeeds
        # vector pins this exactly (a flat two-scalar-edge document must
        # count as one node, not three). This still defends against the
        # YAML alias/anchor amplification the budget exists for (see the
        # docstring above) -- an amplified alias chain still re-walks a
        # real *container* at every occurrence, so it still increments
        # here at the same rate; only scalar leaves stopped counting.
        budget[0] += 1
        if budget[0] > _MAX_NODES:
            raise DocumentError(
                f"{path}: too many nodes materialized (over {_MAX_NODES}) -- "
                "likely a YAML alias/anchor amplification (a shared node "
                "reached via more than one reference is walked in full at "
                "each occurrence)", code="document.limit.nodes", path="$")
        seen = seen or frozenset()
        if id(value) in seen:
            raise DocumentError(f"{path}: cycle detected")
        seen = seen | {id(value)}
        edges: List[Edge] = []
        for k, v in value.items():
            if not isinstance(k, str):
                raise DocumentError(f"{path}: object key {k!r} is not a string",
                                    code="document.unlabeled-element", path=path)
            kp = _join(path, k)
            for child in _children(v, kp, depth + 1, seen, budget):
                edges.append((k, child))
        return edges
    if isinstance(value, (list, tuple)):
        raise DocumentError(f"{path}: a bare array has no labeled-edge form "
                            "(arrays appear only as a repeated field)",
                            code="document.unlabeled-element", path=path)
    if _is_scalar(value):
        _check_int_digits(value, path)
        return value
    raise DocumentError(f"{path}: {type(value).__name__} is not a Document value")

doc(value)

Build a :class:Doc from a plain Python value.

Source code in omnist/document.py

def doc(value: Any) -> Doc:
    """Build a :class:`Doc` from a plain Python value."""
    return value if isinstance(value, Doc) else Doc.of(value)

---

Schema Model (omnist.schema)

The Schema model — two state kinds plus naming, per docs/design/model.md.

• Record — a closed set of fields, each (label, type, cardinality); constrained by its child labels. Cardinality is the unordered number of times a label may appear.
• Scalar — one of exactly seven predefined value types (string, integer, number, boolean, date, time, datetime), optionally nullable. There is no user-declared scalar/value-domain composition — a field's value side is always exactly one of the seven, never a union, an enum, or a literal. (See docs/design/model.md for why: a composable value-domain made schema-directed deserialization ambiguous — a value could satisfy more than one candidate representation with no principled way to choose.)
• Ref — a pointer into the schema's named environment (records only); enables reuse and recursion.

A field's type is a Ref (to a named record) or a Scalar. There are no inline records and no separate array type — "array" is just a field with cardinality max > 1. Validation ignores order.

AnyType

The any type: accepts every legal Document value. Singleton; use t.any. Not a Scalar (it has no kind and no nullable flag — null is already included) and not a Ref (it names nothing).

Source code in omnist/schema.py

class AnyType:
    """The `any` type: accepts every legal Document value. Singleton;
    use ``t.any``. Not a Scalar (it has no kind and no nullable flag —
    null is already included) and not a Ref (it names nothing)."""
    __slots__ = ()
    def __eq__(self, other: object) -> bool: return isinstance(other, AnyType)
    def __hash__(self) -> int: return hash(AnyType)
    def __repr__(self) -> str: return "t.any"

Field

One named, cardinality-bound field slot of a record: label of type, occurring [min, max] times (max=None is unbounded).

Source code in omnist/schema.py

class Field:
    """One named, cardinality-bound field slot of a record: ``label`` of
    ``type``, occurring ``[min, max]`` times (``max=None`` is unbounded)."""

    __slots__ = ("label", "type", "min", "max")

    def __init__(self, label: str, type: Type, min: int = 1,
                 max: Optional[int] = 1) -> None:
        if not isinstance(type, (Ref, Scalar, AnyType)):
            raise SchemaError(
                f"field {label!r} type must be a Ref, Scalar, or t.any, got {type!r}")
        if min < 0 or (max is not None and max < min):
            raise SchemaError(f"field {label!r} has an invalid cardinality [{min},{max}]",
                              code="schema.invalid-cardinality")
        # Note: [0,0] is deliberately still legal to construct directly here
        # (issue #322 only makes it illegal in OSD *text* -- osd.py's
        # _field() -- not as a Python-model invariant; prune()/minimize()
        # construct/consume Field(..., 0, 0) internally as a "dead field"
        # marker, confirmed by tests/test_canonical.py's TestEmptySchemas).
        self.label = label
        self.type = type
        self.min = min
        self.max = max

    def cardinality_str(self) -> str:
        if (self.min, self.max) == (1, 1):
            return "exactly 1"
        if (self.min, self.max) == (0, 1):
            return "0 or 1"
        if self.max is None:
            return f"at least {self.min}"
        return f"between {self.min} and {self.max}"

    def __repr__(self) -> str:
        hi = "" if self.max is None else self.max
        return f"Field({self.label!r}[{self.min},{hi}]: {self.type!r})"

    def __eq__(self, other: Any) -> bool:
        return (isinstance(other, Field) and self.label == other.label
                and self.type == other.type and self.min == other.min
                and self.max == other.max)

Record

A closed set of named fields (constrained by its child labels).

Source code in omnist/schema.py

class Record:
    """A closed set of named fields (constrained by its child labels)."""

    __slots__ = ("fields", "_by_label")

    def __init__(self, fields: List[Field]) -> None:
        self.fields = list(fields)
        self._by_label: Dict[str, Field] = {}
        seen = set()
        for f in self.fields:
            if f.label in seen:
                raise SchemaError(f"duplicate field label {f.label!r} in a record",
                                  code="schema.duplicate-field")
            seen.add(f.label)
            self._by_label[f.label] = f

    def field(self, label: str) -> Optional[Field]:
        return self._by_label.get(label)

    def __repr__(self) -> str:
        return "record{" + ", ".join(repr(f) for f in self.fields) + "}"

    def __eq__(self, other: Any) -> bool:
        # Fields form an unordered set at the model layer (declaration
        # order isn't semantically significant) -- comparing the
        # label-keyed dicts is order-independent for free (dict equality
        # ignores insertion order), and duplicate labels are already
        # rejected by __init__, so this dict is exactly the field set.
        return isinstance(other, Record) and self._by_label == other._by_label

Ref

A reference to a named record.

Source code in omnist/schema.py

class Ref:
    """A reference to a named record."""

    __slots__ = ("name",)

    def __init__(self, name: str) -> None:
        self.name = name

    def __repr__(self) -> str:
        return f"ref({self.name})"

    def __eq__(self, other: Any) -> bool:
        return isinstance(other, Ref) and other.name == self.name

    def __hash__(self) -> int:
        return hash((Ref, self.name))

Scalar

One of the seven predefined value types, optionally nullable.

STRING, INTEGER, … (also under t.*) are ready-to-use, non-nullable instances — a field's type can be one of them directly, with no wrapping needed. Use :func:nullable for the ? form.

Source code in omnist/schema.py

class Scalar:
    """One of the seven predefined value types, optionally nullable.

    ``STRING``, ``INTEGER``, … (also under ``t.*``) are ready-to-use,
    non-nullable instances — a field's type can be one of them directly,
    with no wrapping needed.  Use :func:`nullable` for the ``?`` form.
    """
    __slots__ = ("name", "nullable")

    def __init__(self, name: str, nullable: bool = False) -> None:
        if name not in SCALAR_NAMES:
            raise SchemaError(f"unknown scalar {name!r}; expected one of "
                              f"{sorted(SCALAR_NAMES)}")
        self.name = name
        self.nullable = bool(nullable)

    def __repr__(self) -> str:
        return f"{self.name}{'?' if self.nullable else ''}"

    def __eq__(self, other: Any) -> bool:
        return (isinstance(other, Scalar) and other.name == self.name
                and other.nullable == self.nullable)

    def __hash__(self) -> int:
        return hash((Scalar, self.name, self.nullable))

Schema

A schema: a root reference plus an environment of named records.

Source code in omnist/schema.py

class Schema:
    """A schema: a root reference plus an environment of named records."""

    def __init__(self, root: Ref, env: Optional[Dict[str, Record]] = None) -> None:
        if not isinstance(root, Ref):
            raise SchemaError("a schema root must be a Ref to a named record")
        self.root = root
        self.env: Dict[str, Record] = dict(env or {})
        self.check_refs()

    def resolve(self, t: Type) -> Union[Record, Scalar, AnyType]:
        """An ``AnyType`` or bare ``Scalar`` resolves to itself; a ``Ref`` is a
        single environment lookup -- env values are always Records (enforced
        by ``check_refs``), so ref chains cannot occur."""
        if isinstance(t, AnyType):
            return t
        if isinstance(t, Scalar):
            return t
        if t.name not in self.env:
            raise SchemaError(f"unknown type {t.name!r}", code="schema.unknown-type")
        return self.env[t.name]

    def check_refs(self) -> None:
        for name, rec in self.env.items():
            if not isinstance(rec, Record):
                raise SchemaError(
                    f"environment entry {name!r} must be a Record, got {rec!r}")
            if name in SCALAR_NAMES or name == "any":
                raise SchemaError(
                    f"record name {name!r} shadows a scalar keyword; type "
                    "position resolves a bare name to a builtin first, so "
                    "this record could never be referenced",
                    code="schema.reserved-name", path=name)

        def walk(t: Type, path: str) -> None:
            if isinstance(t, Ref) and t.name not in self.env:
                # E-13: a Schema path -- the field that names the missing
                # record, or `$` when it is the root declaration itself.
                raise SchemaError(f"unknown type {t.name!r}",
                                  code="schema.unknown-type", path=path)
        walk(self.root, "$")
        for name, rec in self.env.items():
            for f in rec.fields:
                walk(f.type, f"{name}.{f.label}")
        # every env value is now known to be a Record (checked above), and
        # root is always a Ref (checked in __init__), so resolve(root) always
        # lands on a Record once the walk above confirms root.name is known.

    # -- validation -----------------------------------------------------
    def validate(self, doc: Any) -> ValidationResult:
        from .document import Doc
        if not isinstance(doc, Doc):
            raise TypeError("validate() expects a Doc; wrap your data with doc(...)")
        res = ValidationResult()
        self._conform(doc, self.root, res, 0)
        return res

    def accepts(self, doc: Any) -> bool:
        return self.validate(doc).ok

    def _conform(self, doc: Any, t: Type, res: ValidationResult, depth: int) -> None:
        from .document import _MAX_DEPTH
        if depth > _MAX_DEPTH:
            raise DocumentError(
                f"{doc.path}: nesting exceeds the maximum depth ({_MAX_DEPTH})")
        d = self.resolve(t)
        if isinstance(d, AnyType):
            return
        if isinstance(d, Scalar):
            self._conform_scalar(doc, d, res)
        else:
            self._conform_record(doc, d, res, depth)

    def _conform_scalar(self, doc: Any, s: Scalar, res: ValidationResult) -> None:
        if not doc.is_leaf:
            res.add(
                doc.path, f"expected a {s.name} value, got an object", "validate.shape-mismatch"
            )
            return
        v = doc.value
        if v is None:
            if not s.nullable:
                res.add(doc.path, "null not allowed here", "validate.null-not-allowed")
            return
        if not matches_kind(v, s.name):
            res.add(
                doc.path, f"expected {s.name}, got {_typename(v)} ({v!r})", "validate.type-mismatch"
            )

    def _conform_record(self, doc: Any, rec: Record, res: ValidationResult,
                        depth: int) -> None:
        if doc.is_leaf:
            res.add(doc.path, "expected an object, got a value", "validate.shape-mismatch")
            return
        counts: Dict[str, int] = {}
        for label, child in doc.edges():
            counts[label] = counts.get(label, 0) + 1
            f = rec.field(label)
            if f is None:
                res.add(child.path, "unexpected field", "validate.unexpected-field")
            else:
                self._conform(child, f.type, res, depth + 1)
        for f in rec.fields:
            c = counts.get(f.label, 0)
            if c < f.min or (f.max is not None and c > f.max):
                res.add(doc.path,
                        f"field {f.label!r} occurs {c} time(s), expected {f.cardinality_str()}",
                        "validate.cardinality")

    # -- comparison (delegate to operations) ----------------------------
    def compatible_with(self, other: "Schema") -> bool:
        """True if every document this schema accepts is also accepted by
        ``other`` (this is a subschema; ``other`` is backward-compatible)."""
        from .ops import compatible_with
        return compatible_with(self, other)

    def equivalent(self, other: "Schema") -> bool:
        """True if both schemas accept exactly the same documents."""
        from .ops import equivalent
        return equivalent(self, other)

    def isomorphic_to(self, other: "Schema") -> bool:
        """Stricter than ``equivalent()`` -- same document language is
        necessary but not sufficient; this additionally requires the same
        record graph structure, up to a renaming of records. Not a
        replacement for ``equivalent()`` as the definition of schema
        equality; used where a caller specifically needs to detect
        structural differences that don't change accepted documents (e.g.
        ``omnist-spec``'s conformance harness checking ``infer``'s output,
        since ``infer`` is documented to never merge duplicate records the
        way ``normalize`` does)."""
        from .ops.isomorphic import _isomorphic
        return _isomorphic(self, other)

    def normalize(self) -> "Schema":
        """The canonical minimal schema equivalent to this one: fewest env
        records, unique up to naming (via partition refinement)."""
        from .ops import normalize
        return normalize(self)

    def is_empty(self) -> bool:
        """True iff this schema's root record is unsatisfiable -- no finite
        document conforms to it (the schema's language is empty)."""
        from .ops import is_empty
        return is_empty(self)

    def prune(self) -> "Schema":
        """An equivalent schema with everything that can never match
        removed: unreachable records, never-emittable (``max == 0``)
        fields, and optional fields whose type can never be satisfied."""
        from .ops import prune
        return prune(self)

    def extract(self, *labels: str) -> "Schema":
        """The minimal subschema that only recognizes documents built from
        ``labels`` (paper Algorithm 5, ExtractSubschema). Fields whose label
        isn't in ``labels`` are dropped; if that deletes a mandatory
        (``min >= 1``) field, the record that had it -- and transitively
        anything that mandatorily depends on it -- is invalidated. Raises
        :class:`~omnist.SchemaError` if the root itself ends up invalidated
        (no valid subschema exists for this label set)."""
        from .ops import extract
        return extract(self, labels)

    # -- serialization --------------------------------------------------
    def to_osd(self, *, indent: Optional[int] = 4) -> str:
        from .osd import to_osd
        return to_osd(self, indent=indent)

    def __repr__(self) -> str:
        return f"Schema(root={self.root!r}, env={list(self.env)})"

    def __eq__(self, other: Any) -> bool:
        # env is a plain dict; dict equality is already order-independent
        # (compares keys and values, ignores insertion order), and
        # declaration order is preserved only for OSD-text readability,
        # not semantically significant -- see Record.__eq__ above.
        return (isinstance(other, Schema) and self.root == other.root
                and self.env == other.env)

compatible_with(other)

True if every document this schema accepts is also accepted by other (this is a subschema; other is backward-compatible).

Source code in omnist/schema.py

def compatible_with(self, other: "Schema") -> bool:
    """True if every document this schema accepts is also accepted by
    ``other`` (this is a subschema; ``other`` is backward-compatible)."""
    from .ops import compatible_with
    return compatible_with(self, other)

equivalent(other)

True if both schemas accept exactly the same documents.

Source code in omnist/schema.py

def equivalent(self, other: "Schema") -> bool:
    """True if both schemas accept exactly the same documents."""
    from .ops import equivalent
    return equivalent(self, other)

extract(*labels)

The minimal subschema that only recognizes documents built from labels (paper Algorithm 5, ExtractSubschema). Fields whose label isn't in labels are dropped; if that deletes a mandatory (min >= 1) field, the record that had it -- and transitively anything that mandatorily depends on it -- is invalidated. Raises :class:~omnist.SchemaError if the root itself ends up invalidated (no valid subschema exists for this label set).

Source code in omnist/schema.py

def extract(self, *labels: str) -> "Schema":
    """The minimal subschema that only recognizes documents built from
    ``labels`` (paper Algorithm 5, ExtractSubschema). Fields whose label
    isn't in ``labels`` are dropped; if that deletes a mandatory
    (``min >= 1``) field, the record that had it -- and transitively
    anything that mandatorily depends on it -- is invalidated. Raises
    :class:`~omnist.SchemaError` if the root itself ends up invalidated
    (no valid subschema exists for this label set)."""
    from .ops import extract
    return extract(self, labels)

is_empty()

True iff this schema's root record is unsatisfiable -- no finite document conforms to it (the schema's language is empty).

Source code in omnist/schema.py

def is_empty(self) -> bool:
    """True iff this schema's root record is unsatisfiable -- no finite
    document conforms to it (the schema's language is empty)."""
    from .ops import is_empty
    return is_empty(self)

isomorphic_to(other)

Stricter than equivalent() -- same document language is necessary but not sufficient; this additionally requires the same record graph structure, up to a renaming of records. Not a replacement for equivalent() as the definition of schema equality; used where a caller specifically needs to detect structural differences that don't change accepted documents (e.g. omnist-spec's conformance harness checking infer's output, since infer is documented to never merge duplicate records the way normalize does).

Source code in omnist/schema.py

def isomorphic_to(self, other: "Schema") -> bool:
    """Stricter than ``equivalent()`` -- same document language is
    necessary but not sufficient; this additionally requires the same
    record graph structure, up to a renaming of records. Not a
    replacement for ``equivalent()`` as the definition of schema
    equality; used where a caller specifically needs to detect
    structural differences that don't change accepted documents (e.g.
    ``omnist-spec``'s conformance harness checking ``infer``'s output,
    since ``infer`` is documented to never merge duplicate records the
    way ``normalize`` does)."""
    from .ops.isomorphic import _isomorphic
    return _isomorphic(self, other)

normalize()

The canonical minimal schema equivalent to this one: fewest env records, unique up to naming (via partition refinement).

Source code in omnist/schema.py

def normalize(self) -> "Schema":
    """The canonical minimal schema equivalent to this one: fewest env
    records, unique up to naming (via partition refinement)."""
    from .ops import normalize
    return normalize(self)

prune()

An equivalent schema with everything that can never match removed: unreachable records, never-emittable (max == 0) fields, and optional fields whose type can never be satisfied.

Source code in omnist/schema.py

def prune(self) -> "Schema":
    """An equivalent schema with everything that can never match
    removed: unreachable records, never-emittable (``max == 0``)
    fields, and optional fields whose type can never be satisfied."""
    from .ops import prune
    return prune(self)

resolve(t)

An AnyType or bare Scalar resolves to itself; a Ref is a single environment lookup -- env values are always Records (enforced by check_refs), so ref chains cannot occur.

Source code in omnist/schema.py

def resolve(self, t: Type) -> Union[Record, Scalar, AnyType]:
    """An ``AnyType`` or bare ``Scalar`` resolves to itself; a ``Ref`` is a
    single environment lookup -- env values are always Records (enforced
    by ``check_refs``), so ref chains cannot occur."""
    if isinstance(t, AnyType):
        return t
    if isinstance(t, Scalar):
        return t
    if t.name not in self.env:
        raise SchemaError(f"unknown type {t.name!r}", code="schema.unknown-type")
    return self.env[t.name]

nullable(scalar)

A copy of scalar that also accepts null (the ? form).

Source code in omnist/schema.py

def nullable(scalar: Scalar) -> Scalar:
    """A copy of ``scalar`` that also accepts ``null`` (the ``?`` form)."""
    if isinstance(scalar, AnyType):
        raise SchemaError("any already includes null; 'any?' is redundant",
                          code="schema.nullable-any")
    if isinstance(scalar, Ref):
        raise SchemaError(
            "nullable() cannot be applied to a Ref; use cardinality [0,1] "
            "for an optional record", code="schema.nullable-ref")
    return scalar if scalar.nullable else Scalar(scalar.name, True)

value_kind(v)

The most specific scalar name a Python value matches, for inference and error messages (integer is reported even though it also matches number — callers needing the wider check use :func:matches_kind).

Source code in omnist/schema.py

def value_kind(v: Any) -> str:
    """The most specific scalar name a Python value matches, for inference
    and error messages (``integer`` is reported even though it also matches
    ``number`` — callers needing the wider check use :func:`matches_kind`)."""
    if isinstance(v, bool):
        return "boolean"
    if isinstance(v, int):
        return "integer"
    if isinstance(v, float):
        return "number"
    if isinstance(v, _dt.datetime):
        return "datetime"
    if isinstance(v, _dt.date):
        return "date"
    if isinstance(v, _dt.time):
        return "time"
    return "string"

---

OSD Grammar & Serialization (omnist.osd)

OSD (Omnist Schema Definition) — the text language for the Schema model.

Grammar (informal)::

schema      := record* 'root' NAME
record      := 'record' NAME '{' field (',' field)* ','? '}'
field       := STRING cardinality? ':' type
cardinality := '[' INT? (',' INT?)? ']'          -- [m,n] [m,] [,n] [n]; absent = [1,1]
type        := SCALARNAME '?'? | NAME            -- one scalar, or one Ref

Quoting rule: a "quoted" token is a data string (always a field label — there is no other use for a string literal in this grammar); an unquoted identifier is a schema name (a scalar keyword, or a Ref).

There is no value-domain composition: no |, no enum, no literal-valued fields, and no union/domain declaration. A field's type is always either one of the seven scalars (string, integer, number, boolean, date, time, datetime), optionally ?, or a Ref to a named record. See docs/design/model.md for why: a composable value-domain made schema-directed deserialization ambiguous.

parse_schema(text)

Parse OSD text into a :class:~omnist.schema.Schema.

Source code in omnist/osd.py

def parse_schema(text: str) -> Schema:
    """Parse OSD text into a :class:`~omnist.schema.Schema`."""
    text = strip_bom(text)   # Sec2.5 D-15; a second mark fails as a stray character (D-21)
    return _Parser(_tokenize(text), text).parse()

to_osd(schema, *, indent=4)

Serialize a Schema back to OSD text.

indent=None renders a single-line, machine-oriented form (record defs and the root statement joined by spaces, fields joined by ", ", no trailing comma) instead of the default pretty-printed, indented form -- mirroring write_oml/write_json's own indent=None convention. A non-None int sets the pretty-mode indent width (default 4, matching the prior hardcoded behavior). Both forms round-trip through parse_schema.

A field label is written between double quotes with exactly two escapes (Sec5.9, OSD-15): a backslash is doubled and a double quote gets a backslash in front of it, nothing else -- OSD's unescaping is weak, so those two are all that is needed for every label to read back as itself. A label with a C0 control character (below U+0020) has no OSD spelling at all, so writing it raises :class:~omnist.errors.WriteError with code="write.unsupported-value" and path the Schema path of the record holding the field (OSD-14, E-26), unconditionally.

Source code in omnist/osd.py

def to_osd(schema: Schema, *, indent: Optional[int] = 4) -> str:
    """Serialize a Schema back to OSD text.

    ``indent=None`` renders a single-line, machine-oriented form (record
    defs and the ``root`` statement joined by spaces, fields joined by
    ``", "``, no trailing comma) instead of the default pretty-printed,
    indented form -- mirroring ``write_oml``/``write_json``'s own
    ``indent=None`` convention. A non-``None`` int sets the pretty-mode
    indent width (default 4, matching the prior hardcoded behavior). Both
    forms round-trip through ``parse_schema``.

    A field label is written between double quotes with exactly two
    escapes (Sec5.9, OSD-15): a backslash is doubled and a double quote gets
    a backslash in front of it, nothing else -- OSD's unescaping is weak, so
    those two are all that is needed for every label to read back as itself.
    A label with a C0
    control character (below ``U+0020``) has no OSD spelling at all, so
    writing it raises :class:`~omnist.errors.WriteError` with
    ``code="write.unsupported-value"`` and ``path`` the Schema path of the
    *record* holding the field (OSD-14, E-26), unconditionally.
    """
    parts: List[str] = [_record(name, rec, indent) for name, rec in schema.env.items()]
    parts.append(f"root {schema.root.name}")
    if indent is None:
        return " ".join(parts) + "\n"
    return "\n".join(parts) + "\n"

---

Codecs & Formats (omnist.formats & omnist.oml)

Codecs over the canonical Document (edge-list) model.

Readers parse a format into a node; writers project a node back. JSON/YAML/TOML go through the JSON-shaped grouping (to_grouped); XML uses repeated elements directly, so it preserves interleaving on read and needs a single document element on write.

Writing is lenient by default: when a value can't be held losslessly (TOML has no null; JSON/XML have no date type), the writer adjusts it and records the change in a :class:~omnist.report.WriteReport. Pass report= to inspect, or strict=True to raise on any adjustment. See :mod:~omnist.report.

check_json(node)

Report what writing JSON would adjust, without producing output.

Source code in omnist/formats.py

def check_json(node: Any) -> WriteReport:
    """Report what writing JSON would adjust, without producing output."""
    return _scan_json(node)

OML (Omnist Markup Language) — the native codec for the Document model.

OML is omnist's own serialization format: every Document — every ordered, possibly-repeated, possibly-interleaved edge list, and all seven scalar kinds (string, integer, number, boolean, date, time, datetime) plus null — round-trips through OML exactly, with no adjustment ever needed (unlike JSON/YAML/TOML/XML, OML never has a :class:~omnist.report.WriteReport entry to report).

This module implements the OML-Core grammar in full, plus the OML-Extended raw-string and triple-quoted multiline-string spellings (E2/E3) on read. The canonical writer only ever emits OML-Core.

See docs/formats/oml.md for the user-facing guide and docs/design/OML-spec.md (design-time artifact, not shipped) for the full normative grammar this implementation follows.

Performance note (issue #168): the reader is a single-pass scanner/parser built around one compiled "master" regex with named groups. There is no Token class and no materialized token list — the parser drives a master.match(s, pos) loop directly off Match objects and dispatches on m.lastgroup. Per-token line/col is not computed during scanning; it's derived lazily, only when a ParseError is actually raised, by counting newlines in s[:pos]. Scalar values (int()/float()/date parsing) are likewise computed only when a token is consumed by the parser, not when it's scanned. This is what makes the single-pass design pay off: per-token Python-level object construction was the dominant cost, not regex matching itself (see the PR for the profile that motivated this).

check_oml(node)

OML can hold every Document losslessly; always an empty report.

Source code in omnist/oml.py

def check_oml(node: Any) -> "WriteReport":
    """OML can hold every Document losslessly; always an empty report."""
    from .report import WriteReport
    return WriteReport()

read_oml(text, *, schema=None)

Parse OML source into a canonical Document node (edge-list or leaf).

Source code in omnist/oml.py

def read_oml(text: str, *, schema: Optional[Any] = None) -> Any:
    """Parse OML source into a canonical Document node (edge-list or leaf)."""
    scanner = _Scanner(strip_bom(text))   # Sec2.5 D-15
    node = _Parser(scanner).parse_document()
    if schema is None:
        return node
    from .deserialize import materialize
    return materialize(node, schema)

write_oml(node, *, indent=2, arrays=False)

Render a canonical Document node as OML source.

OML is lossless for every Document: there is never an adjustment to report (unlike JSON/YAML/TOML/XML), so there is no check_oml/ strict=/report= machinery — the write always succeeds exactly.

indent=None renders a single-line, machine-oriented form (edges joined by "; ", no newlines/padding) instead of the default pretty-printed, indented form -- mirroring write_json's own indent=None convention. Both forms round-trip through read_oml.

arrays=True (issue #218) collapses any maximal run of >= 2 consecutive same-label edges into label: [v1, v2, ...] array syntax -- a run of length 1 still writes as a plain scalar edge, and a run is never merged across an edge with a different label in between, so this never reorders anything: read_oml(write_oml(node, arrays=True)) == node holds unconditionally. Default False produces output byte-identical to arrays not existing at all.

Source code in omnist/oml.py

def write_oml(node: Any, *, indent: Optional[int] = 2, arrays: bool = False) -> str:
    """Render a canonical Document node as OML source.

    OML is lossless for every Document: there is never an adjustment to
    report (unlike JSON/YAML/TOML/XML), so there is no ``check_oml``/
    ``strict=``/``report=`` machinery — the write always succeeds exactly.

    ``indent=None`` renders a single-line, machine-oriented form (edges
    joined by ``"; "``, no newlines/padding) instead of the default
    pretty-printed, indented form -- mirroring ``write_json``'s own
    ``indent=None`` convention. Both forms round-trip through ``read_oml``.

    ``arrays=True`` (issue #218) collapses any maximal run of >= 2
    consecutive same-label edges into ``label: [v1, v2, ...]`` array
    syntax -- a run of length 1 still writes as a plain scalar edge, and a
    run is never merged across an edge with a different label in between,
    so this never reorders anything: ``read_oml(write_oml(node,
    arrays=True)) == node`` holds unconditionally. Default ``False``
    produces output byte-identical to ``arrays`` not existing at all.
    """
    if not isinstance(node, list):
        return _write_scalar(node)
    if indent is None:
        return _write_edges_compact(node, arrays, 0)
    return _write_edges(node, 0, indent, arrays, 0)

---

Inference & Materialization (omnist.infer & omnist.deserialize)

Infer a Schema from example Documents, on the canonical model.

Given one or more sample Documents, draft a record schema that accepts them:

• a label present in every sample with count 1 becomes a required field ([1,1]); absent in some samples -> [0,1]; seen more than once -> an array ([min,]);
• scalar children become one :class:~omnist.schema.Scalar (nullable if any sample was null). Samples disagreeing on scalar shape raise, except integer/number mixing, which collapses to number (the one subset relation between scalars) -- see docs/design/model.md;
• object children become a nested, named record (recursively).

Since the model has no inline records, nested records are given generated names derived from their label.

infer deliberately does not auto-normalize: the raw result keeps a 1:1 correspondence between sample labels and generated record names, which is easier to read and hand-edit, and may therefore contain structurally- identical duplicate records. Call .normalize() on the result where a canonical minimal schema is wanted (decided in issues #143/#151).

AnyFallback dataclass

A single field infer opened as any under allow_any.

location reads RecordName.label; reason says why the field could not be given a single precise type.

Source code in omnist/infer.py

@dataclass(frozen=True)
class AnyFallback:
    """A single field ``infer`` opened as ``any`` under ``allow_any``.

    ``location`` reads ``RecordName.label``; ``reason`` says why the field
    could not be given a single precise type.
    """

    location: str
    reason: str

Schema-directed deserialization: make a freshly-read node conform to a :class:~omnist.schema.Schema, or raise.

Readers (read_json, etc.) hand back text-shaped values: JSON/YAML/TOML have no date/time type, so a temporal field arrives as an ISO-8601 string; a whole-number float may need to become an int (or vice versa) to match what the schema declares. Passing schema= to a reader is the request for a Document that's guaranteed to conform to that schema: :func:materialize walks the node together with the schema, upgrading each leaf only when the conversion is value-exact -- "2024-01-01" -> date, 1.0 -> int 1 -- and checking every record's shape (closed fields, cardinality) along the way, exactly as :meth:Schema.validate would. If anything can't be made to conform -- an inexact scalar, an unknown field, a missing field, the wrong cardinality -- :func:materialize collects every such problem (not just the first) and raises one :class:~omnist.errors.ParseError with the full report, both as a message string and structurally on .errors (a list of (path, message, code)).

This can't simply delegate to :meth:Schema.validate after the fact: validate only ever checks a value already in its final form, with no notion of upgrading, and it would mean a second, redundant top-down walk of the same tree using different traversal code. Since :func:materialize already knows, at every node, exactly which field/type the schema expects there, upgrading and shape-checking happen together in one pass.

There's no strict= switch: a schema is either given, in which case the result is guaranteed to conform (or an error is raised), or it isn't, in which case the node is returned exactly as read, untouched -- schema=None is the existing, well-defined way to opt out of validation entirely.

materialize(node, schema)

A copy of node with leaf values upgraded to match schema, guaranteed to conform to it -- raises :class:~omnist.errors.ParseError (with every problem found, not just the first, in both the message and the structured .errors list) if it can't be made to.

Source code in omnist/deserialize.py

def materialize(node: Any, schema: Schema) -> Any:
    """A copy of ``node`` with leaf values upgraded to match ``schema``,
    guaranteed to conform to it -- raises :class:`~omnist.errors.ParseError`
    (with every problem found, not just the first, in both the message and
    the structured ``.errors`` list) if it can't be made to."""
    res = ValidationResult()
    out = _materialize_type(node, schema, schema.root, "$", res)
    if not res.ok:
        raise ParseError(str(res), errors=res.errors)
    return out

---

Operations & Linting (omnist.ops)

Schema operations package.

One module per algorithm from the paper (Lee & Cheung, "XML Schema Computations", CIKM 2010).

LintFinding dataclass

One structural diagnostic. code is a stable machine-readable identifier (lint.unsatisfiable-record, lint.unreachable-record, lint.duplicate-record, lint.any-field); severity is warning or info; location is a record name (or Record.label for lint.any-field); message is a human-readable, actionable description.

Source code in omnist/ops/lint.py

@dataclass(frozen=True)
class LintFinding:
    """One structural diagnostic. ``code`` is a stable machine-readable
    identifier (``lint.unsatisfiable-record``, ``lint.unreachable-record``,
    ``lint.duplicate-record``, ``lint.any-field``); ``severity`` is ``warning`` or
    ``info``; ``location`` is a record name (or ``Record.label`` for
    ``lint.any-field``); ``message`` is a human-readable, actionable description."""

    code: str
    severity: str
    location: str
    message: str

compatible_with(a, b)

True if every document a accepts is also accepted by b (a is a subschema / b is backward-compatible).

Source code in omnist/ops/subschema.py

def compatible_with(a: Schema, b: Schema) -> bool:
    """True if every document ``a`` accepts is also accepted by ``b``
    (``a`` is a subschema / ``b`` is backward-compatible)."""
    sat_a = satisfiable_set(a)
    return _sub(a, a.root, b, b.root, sat_a, {})

equivalence_classes(s)

Partition s.env's record names into structural-equivalence classes via MinimizeSA-style partition refinement (module docstring, steps 2-3): an initial local_signature grouping refined to a fixpoint by which block each same-labeled ref field points to.

Operates on s.env exactly as given -- it does not prune first, so unreachable or unsatisfiable records are still classified. normalize calls this after its own prune/is_empty steps; lint calls it on the raw schema so structurally-identical records are reported as authored. Each returned block is a list of names; a block of length > 1 is a set of records with identical structure.

Source code in omnist/ops/minimize.py

def equivalence_classes(s: Schema) -> List[List[str]]:
    """Partition ``s.env``'s record names into structural-equivalence classes
    via MinimizeSA-style partition refinement (module docstring, steps 2-3):
    an initial ``local_signature`` grouping refined to a fixpoint by which
    *block* each same-labeled ref field points to.

    Operates on ``s.env`` exactly as given -- it does **not** prune first, so
    unreachable or unsatisfiable records are still classified. ``normalize``
    calls this after its own ``prune``/``is_empty`` steps; ``lint`` calls it on
    the raw schema so structurally-identical records are reported as authored.
    Each returned block is a list of names; a block of length > 1 is a set of
    records with identical structure.
    """
    names = sorted(s.env)
    block_of: Dict[str, int] = {}
    blocks: List[List[str]] = _group_by(names, lambda n: local_signature(s.env[n]))
    for i, block in enumerate(blocks):
        for n in block:
            block_of[n] = i

    changed = True
    while changed:
        changed = False
        new_blocks: List[List[str]] = []
        new_block_of: Dict[str, int] = {}
        for block in blocks:
            for sub in _group_by(block, lambda n: _refine_key(s.env[n], block_of)):
                idx = len(new_blocks)
                new_blocks.append(sub)
                for n in sub:
                    new_block_of[n] = idx
        if len(new_blocks) != len(blocks):
            changed = True
        blocks = new_blocks
        block_of = new_block_of
    return blocks

equivalent(a, b)

True if both schemas accept exactly the same documents.

Source code in omnist/ops/subschema.py

def equivalent(a: Schema, b: Schema) -> bool:
    """True if both schemas accept exactly the same documents."""
    return compatible_with(a, b) and compatible_with(b, a)

is_empty(s)

True iff s's root record is unsatisfiable -- the schema's language (the set of documents it accepts) is empty.

Source code in omnist/ops/prune.py

def is_empty(s: Schema) -> bool:
    """True iff ``s``'s root record is unsatisfiable -- the schema's
    language (the set of documents it accepts) is empty."""
    return s.root.name not in satisfiable_set(s)

normalize(s)

The canonical minimal schema equivalent to s: fewest env records, unique up to record naming. See module docstring for the algorithm (paper's Algorithm 2, MinimizeSA).

Source code in omnist/ops/minimize.py

def normalize(s: Schema) -> Schema:
    """The canonical minimal schema equivalent to ``s``: fewest env
    records, unique up to record naming. See module docstring for the
    algorithm (paper's Algorithm 2, MinimizeSA)."""
    s = prune(s)
    if is_empty(s):
        return s

    names = sorted(s.env)
    blocks = equivalence_classes(s)

    rep: Dict[str, str] = {}
    for block in blocks:
        keep = min(block)
        for n in block:
            rep[n] = keep

    new_env: Dict[str, Record] = {}
    for name in names:
        if rep[name] == name:
            new_env[name] = _remap(s.env[name], rep)
    new_root = Ref(rep.get(s.root.name, s.root.name))
    return Schema(new_root, new_env)

satisfiable_set(s)

The set of env record names that admit at least one finite document.

Least fixpoint: start with nothing known-satisfiable and repeatedly add any record all of whose mandatory (min >= 1) fields are already satisfiable (a bare Scalar, or a Ref to an already-satisfiable record). Monotonic on a finite env, so this always terminates.

Source code in omnist/ops/prune.py

def satisfiable_set(s: Schema) -> Set[str]:
    """The set of env record names that admit at least one finite document.

    Least fixpoint: start with nothing known-satisfiable and repeatedly add
    any record all of whose mandatory (``min >= 1``) fields are already
    satisfiable (a bare ``Scalar``, or a ``Ref`` to an already-satisfiable
    record). Monotonic on a finite env, so this always terminates.
    """
    sat: Set[str] = set()
    changed = True
    while changed:
        changed = False
        for name, rec in s.env.items():
            if name in sat:
                continue
            if _record_satisfiable(rec, sat):
                sat.add(name)
                changed = True
    return sat

Non-destructive structural diagnostics for a schema (omnist schema lint).

validate checks a document against a schema; lint checks the schema itself for structural problems that parse fine but mean parts of the schema can never do anything. It reports, never mutates -- that line is the whole design. prune and normalize are the transforms that fix these issues; lint only diagnoses them.

Four checks:

• lint.unsatisfiable-record (warning) -- a reachable record no finite document can match (e.g. a mandatory ref cycle). Reuses :func:prune.satisfiable_set (its complement), intersected with reachable.
• lint.unreachable-record (warning) -- a record defined in env but not reachable from root by following any ref. A plain reachability walk (no pruning): every Ref-typed field is followed regardless of cardinality.
• lint.duplicate-record (warning) -- two or more structurally identical records under different names. Reuses :func:minimize.equivalence_classes on the raw schema, so duplicates are reported as authored.
• lint.any-field (info) -- an inventory of every any-typed field, so a human can audit the schema's deliberate openings. Advisory only; never fails the exit code on its own.

LintFinding dataclass

One structural diagnostic. code is a stable machine-readable identifier (lint.unsatisfiable-record, lint.unreachable-record, lint.duplicate-record, lint.any-field); severity is warning or info; location is a record name (or Record.label for lint.any-field); message is a human-readable, actionable description.

Source code in omnist/ops/lint.py

@dataclass(frozen=True)
class LintFinding:
    """One structural diagnostic. ``code`` is a stable machine-readable
    identifier (``lint.unsatisfiable-record``, ``lint.unreachable-record``,
    ``lint.duplicate-record``, ``lint.any-field``); ``severity`` is ``warning`` or
    ``info``; ``location`` is a record name (or ``Record.label`` for
    ``lint.any-field``); ``message`` is a human-readable, actionable description."""

    code: str
    severity: str
    location: str
    message: str

lint(s)

Structural diagnostics for s -- see the module docstring for the four checks. Returns findings sorted deterministically by (code, location). Never mutates s.

Source code in omnist/ops/lint.py

def lint(s: Schema) -> List[LintFinding]:
    """Structural diagnostics for ``s`` -- see the module docstring for the
    four checks. Returns findings sorted deterministically by ``(code,
    location)``. Never mutates ``s``."""
    findings: List[LintFinding] = []

    reachable = _reachable(s)
    sat = satisfiable_set(s)

    # lint.unsatisfiable-record: reachable but not satisfiable
    for name in reachable - sat:
        findings.append(LintFinding(
            "lint.unsatisfiable-record", "warning", name,
            f"record {name!r} is reachable but unsatisfiable -- no finite "
            f"document can match it (e.g. a mandatory ref cycle)"))

    # lint.unreachable-record: defined in env but not reachable from root
    for name in set(s.env) - reachable:
        findings.append(LintFinding(
            "lint.unreachable-record", "warning", name,
            f"record {name!r} is defined but never reachable from the root; "
            f"drop it with `schema prune`"))

    # lint.duplicate-record: structurally identical records under different names
    for block in equivalence_classes(s):
        if len(block) > 1:
            group = sorted(block)
            location = ", ".join(group)
            keep = group[0]
            others = ", ".join(repr(n) for n in group[1:])
            findings.append(LintFinding(
                "lint.duplicate-record", "warning", location,
                f"records {others} are structurally identical to {keep!r}; "
                f"merge them with `schema normalize`"))

    # lint.any-field: inventory of every any-typed field
    for name in sorted(s.env):
        for f in s.env[name].fields:
            if isinstance(f.type, AnyType):
                findings.append(LintFinding(
                    "lint.any-field", "info", f"{name}.{f.label}",
                    f"field {f.label!r} of record {name!r} is typed `any` "
                    f"(accepts any value unchecked)"))

    findings.sort(key=lambda x: (x.code, x.location))
    return findings

Schema minimization: partition-refinement to the canonical minimal form.

Implements the paper's Algorithm 2 (MinimizeSA) -- the same family as DFA minimization by partition refinement (Hopcroft/Moore-style state merging). normalize(s) returns an equivalent schema with the fewest possible env records, unique up to record naming (paper Theorems 3-4; they transfer to omnist's deterministic counting-language restriction -- see docs/design/model.md).

Algorithm:

1. s = prune(s) -- mandatory first step. Two semantically-equal records must not be kept apart by never-emittable fields or unreachable records; pruning first is what makes the partition canonical (see ops/prune.py).
2. Initial partition: env records grouped by local_signature (see ops/signature.py) -- a target-blind structural key, so records that might turn out equivalent via differently-named ref targets still start in the same block.
3. Refine: split any block whose members disagree, for some label, on which block their same-labeled ref-typed field points to. Repeat until no block splits (a fixpoint -- always reached on a finite env). This is exactly DFA-minimization-style refinement: two states are equivalent iff every transition leads to equivalent states.
4. Merge: collapse each stable block to a single representative -- its lexicographically smallest member name (deterministic) -- and remap every ref and the root to representatives.

Special case: an unsatisfiable (empty-language) root. prune() deliberately leaves such a root's fields untouched (see its docstring), so partition refinement over the unsatisfiable core isn't meaningful -- there's no "fewest records" notion to compute when the schema accepts no finite document at all. In that case normalize just returns the pruned schema unchanged.

equivalence_classes(s)

Partition s.env's record names into structural-equivalence classes via MinimizeSA-style partition refinement (module docstring, steps 2-3): an initial local_signature grouping refined to a fixpoint by which block each same-labeled ref field points to.

Operates on s.env exactly as given -- it does not prune first, so unreachable or unsatisfiable records are still classified. normalize calls this after its own prune/is_empty steps; lint calls it on the raw schema so structurally-identical records are reported as authored. Each returned block is a list of names; a block of length > 1 is a set of records with identical structure.

Source code in omnist/ops/minimize.py

def equivalence_classes(s: Schema) -> List[List[str]]:
    """Partition ``s.env``'s record names into structural-equivalence classes
    via MinimizeSA-style partition refinement (module docstring, steps 2-3):
    an initial ``local_signature`` grouping refined to a fixpoint by which
    *block* each same-labeled ref field points to.

    Operates on ``s.env`` exactly as given -- it does **not** prune first, so
    unreachable or unsatisfiable records are still classified. ``normalize``
    calls this after its own ``prune``/``is_empty`` steps; ``lint`` calls it on
    the raw schema so structurally-identical records are reported as authored.
    Each returned block is a list of names; a block of length > 1 is a set of
    records with identical structure.
    """
    names = sorted(s.env)
    block_of: Dict[str, int] = {}
    blocks: List[List[str]] = _group_by(names, lambda n: local_signature(s.env[n]))
    for i, block in enumerate(blocks):
        for n in block:
            block_of[n] = i

    changed = True
    while changed:
        changed = False
        new_blocks: List[List[str]] = []
        new_block_of: Dict[str, int] = {}
        for block in blocks:
            for sub in _group_by(block, lambda n: _refine_key(s.env[n], block_of)):
                idx = len(new_blocks)
                new_blocks.append(sub)
                for n in sub:
                    new_block_of[n] = idx
        if len(new_blocks) != len(blocks):
            changed = True
        blocks = new_blocks
        block_of = new_block_of
    return blocks

normalize(s)

The canonical minimal schema equivalent to s: fewest env records, unique up to record naming. See module docstring for the algorithm (paper's Algorithm 2, MinimizeSA).

Source code in omnist/ops/minimize.py

def normalize(s: Schema) -> Schema:
    """The canonical minimal schema equivalent to ``s``: fewest env
    records, unique up to record naming. See module docstring for the
    algorithm (paper's Algorithm 2, MinimizeSA)."""
    s = prune(s)
    if is_empty(s):
        return s

    names = sorted(s.env)
    blocks = equivalence_classes(s)

    rep: Dict[str, str] = {}
    for block in blocks:
        keep = min(block)
        for n in block:
            rep[n] = keep

    new_env: Dict[str, Record] = {}
    for name in names:
        if rep[name] == name:
            new_env[name] = _remap(s.env[name], rep)
    new_root = Ref(rep.get(s.root.name, s.root.name))
    return Schema(new_root, new_env)

Satisfiability analysis and schema pruning.

Implements the paper's "useless-state removal" (MakeUsefulSA) analog: a record is satisfiable iff it admits at least one finite document, and :func:prune returns an equivalent schema with everything that can never match removed. This is the precondition Algorithm 4 (SubschemaSA, ops/subschema.py) needs to be correct — see docs/design/model.md for the full satisfiability subsection.

Satisfiability is a least fixpoint over the env's records: a record is satisfiable iff every field with min >= 1 is either a Scalar or a Ref to a satisfiable record. (Fields with min == 0 never block satisfiability -- they simply need not be emitted.) Scalars are always satisfiable, so a record with no mandatory fields at all is trivially satisfiable (the empty document for that record admits it).

is_empty(s)

True iff s's root record is unsatisfiable -- the schema's language (the set of documents it accepts) is empty.

Source code in omnist/ops/prune.py

def is_empty(s: Schema) -> bool:
    """True iff ``s``'s root record is unsatisfiable -- the schema's
    language (the set of documents it accepts) is empty."""
    return s.root.name not in satisfiable_set(s)

prune(s)

An equivalent schema with everything that can never match removed:

• records unreachable from root (following refs) are dropped;
• fields with max == 0 are dropped (never emittable);
• optional (min == 0) fields whose type is an unsatisfiable record are dropped (they could never actually be emitted either);
• records left unreachable/unsatisfiable after the above are dropped from the environment too.

Root-unsatisfiable case. If the root record itself is unsatisfiable (every finite document is rejected -- is_empty() is True), field pruning is not applied to the root: its mandatory fields are exactly what make it unsatisfiable, and stripping them would silently produce a different, satisfiable schema, contradicting "prune returns an equivalent schema." Instead the root record is kept as-is and only the rest of the environment is reduced to what's reachable from it (which, being unsatisfiable, typically collapses to the cyclic core itself). This mirrors the paper's treatment of an unsatisfiable start state -- MakeUsefulSA modifies the automaton rather than rejecting it outright.

Source code in omnist/ops/prune.py

def prune(s: Schema) -> Schema:
    """An equivalent schema with everything that can never match removed:

    * records unreachable from root (following refs) are dropped;
    * fields with ``max == 0`` are dropped (never emittable);
    * optional (``min == 0``) fields whose type is an unsatisfiable record
      are dropped (they could never actually be emitted either);
    * records left unreachable/unsatisfiable after the above are dropped
      from the environment too.

    **Root-unsatisfiable case.** If the root record itself is unsatisfiable
    (every finite document is rejected -- ``is_empty()`` is True), field
    pruning is *not* applied to the root: its mandatory fields are exactly
    what make it unsatisfiable, and stripping them would silently produce a
    *different*, satisfiable schema, contradicting "prune returns an
    equivalent schema." Instead the root record is kept as-is and only the
    rest of the environment is reduced to what's reachable from it (which,
    being unsatisfiable, typically collapses to the cyclic core itself).
    This mirrors the paper's treatment of an unsatisfiable start state --
    MakeUsefulSA modifies the automaton rather than rejecting it outright.
    """
    sat = satisfiable_set(s)
    root_ok = s.root.name in sat

    reachable = _reachable(s, sat, root_ok)

    # Iterate s.env (a dict, insertion-order-stable regardless of hash
    # seed) filtered to `reachable` (a set, whose own iteration order is
    # PYTHONHASHSEED-dependent for str keys) -- not the other way round.
    # Otherwise new_env's key order, and so prune()'s output env order,
    # varies nondeterministically across process runs (issue #253).
    new_env: Dict[str, Record] = {}
    for name in s.env:
        if name not in reachable:
            continue
        rec = s.env[name]
        if not root_ok and name == s.root.name:
            new_env[name] = rec           # keep the unsatisfiable root intact
        else:
            new_env[name] = _prune_record(rec, sat)
    return Schema(Ref(s.root.name), new_env)

satisfiable_set(s)

The set of env record names that admit at least one finite document.

Least fixpoint: start with nothing known-satisfiable and repeatedly add any record all of whose mandatory (min >= 1) fields are already satisfiable (a bare Scalar, or a Ref to an already-satisfiable record). Monotonic on a finite env, so this always terminates.

Source code in omnist/ops/prune.py

def satisfiable_set(s: Schema) -> Set[str]:
    """The set of env record names that admit at least one finite document.

    Least fixpoint: start with nothing known-satisfiable and repeatedly add
    any record all of whose mandatory (``min >= 1``) fields are already
    satisfiable (a bare ``Scalar``, or a ``Ref`` to an already-satisfiable
    record). Monotonic on a finite env, so this always terminates.
    """
    sat: Set[str] = set()
    changed = True
    while changed:
        changed = False
        for name, rec in s.env.items():
            if name in sat:
                continue
            if _record_satisfiable(rec, sat):
                sat.add(name)
                changed = True
    return sat

Subschema compatibility and equivalence.

Implements the paper's Algorithm 4 (SubschemaSA) restricted to omnist's counting cardinality languages; equivalent is bidirectional inclusion.

Algorithm 4 assumes its precondition MakeUsefulSA (useless-state removal, ops/prune.py) has already run: the coinductive cycle rule below only coincides with true (finite-document) language inclusion once every A-side record is known satisfiable. Rather than requiring callers to pre-prune, compatible_with computes a's satisfiable set once up front and _sub/_record_sub consult it directly -- an unsatisfiable A-side record is vacuously a subschema of anything (it emits no documents at all), and an optional A-field whose type is unsatisfiable is skipped (it can never actually be emitted, so it imposes no obligation on B). See docs/design/model.md for the full satisfiability subsection.

compatible_with(a, b)

True if every document a accepts is also accepted by b (a is a subschema / b is backward-compatible).

Source code in omnist/ops/subschema.py

def compatible_with(a: Schema, b: Schema) -> bool:
    """True if every document ``a`` accepts is also accepted by ``b``
    (``a`` is a subschema / ``b`` is backward-compatible)."""
    sat_a = satisfiable_set(a)
    return _sub(a, a.root, b, b.root, sat_a, {})

equivalent(a, b)

True if both schemas accept exactly the same documents.

Source code in omnist/ops/subschema.py

def equivalent(a: Schema, b: Schema) -> bool:
    """True if both schemas accept exactly the same documents."""
    return compatible_with(a, b) and compatible_with(b, a)

Subschema extraction (paper Algorithm 5, ExtractSubschema).

Given a schema and a set of permissible labels keep (the paper's X'), produces the minimal subschema that recognizes only documents built from those labels -- the headline application in the paper is trimming a large shared schema (xCBL) down to just what a single document type needs (reported there as a 6-32% size reduction).

Algorithm:

1. For every record in the env, delete any field whose label is not in keep.
2. If a deleted field had min >= 1 (mandatory), that record is invalidated -- the paper's "state removed": there is no way to build a document at that record's shape without a label that's no longer available, so the record itself can no longer be produced.
3. Propagate. A record with a mandatory field whose type is an invalidated record is itself invalidated (that field can never be filled), and so on transitively -- a least-fixpoint closure, same shape as ops/prune.py's satisfiability fixpoint.
4. If the root ends up invalidated, there is no valid subschema for this keep set at all: :func:extract raises :class:~omnist.SchemaError naming the first offending label and record, so the failure is actionable.
5. Otherwise, invalidated records (and fields typed to them, along with any fields already dropped in step 1) are gone; the result is run through :func:~omnist.ops.prune.prune and :func:~omnist.ops.minimize.normalize (Algorithm 5's own final MakeUseful + Minimize step) to land in the same canonical minimal form normalize() produces elsewhere.

Design decision: mandatory deletion is an error, not silently-optional. An alternative design could relax a deleted mandatory field to optional instead of invalidating its record. This implementation deliberately does not do that: silently loosening cardinality would mean extract's result no longer reflects the paper's Algorithm 5 semantics (which reports "no valid subschema" rather than inventing a weaker one), and it would hide a likely mistake -- asking to keep a leaf label without any of the mandatory structure that leads to it is far more often a bug in the caller's keep set than an intentional relaxation. Callers who do want the relaxed behavior can trivially get it by editing field cardinalities before calling extract.

extract(s, keep)

The minimal subschema of s that only recognizes documents built from labels in keep. Raises :class:SchemaError if deleting the other labels would invalidate the root record (see module docstring).

Source code in omnist/ops/extract.py

def extract(s: Schema, keep: Iterable[str]) -> Schema:
    """The minimal subschema of ``s`` that only recognizes documents built
    from labels in ``keep``. Raises :class:`SchemaError` if deleting the
    other labels would invalidate the root record (see module docstring)."""
    keep_set: Set[str] = set(keep)

    # Step 1+2: per-record field deletion, tracking which records are
    # directly invalidated by the loss of a mandatory field, and the first
    # offending (label, record) pair for the error message.
    trimmed: Dict[str, Record] = {}
    invalidated: Set[str] = set()
    first_offender: Optional[tuple[str, str]] = None

    for name, rec in s.env.items():
        kept_fields = []
        for f in rec.fields:
            if f.label in keep_set:
                kept_fields.append(f)
            elif f.min >= 1:
                if name not in invalidated and first_offender is None:
                    first_offender = (f.label, name)
                invalidated.add(name)
        trimmed[name] = Record(kept_fields)

    # Step 3: propagate invalidation -- a record with a mandatory field
    # typed to an invalidated record is itself invalidated. Least fixpoint,
    # same shape as prune.py's satisfiable_set.
    changed = True
    while changed:
        changed = False
        for name, rec in trimmed.items():
            if name in invalidated:
                continue
            for f in rec.fields:
                if f.min >= 1 and isinstance(f.type, Ref) and f.type.name in invalidated:
                    # first_offender is always already set here: this branch
                    # can only run once `invalidated` is non-empty, and it's
                    # only ever seeded by step 1, which sets first_offender
                    # itself before propagation ever begins.
                    invalidated.add(name)
                    changed = True
                    break

    # Step 4: root invalidated -> no valid subschema.
    if s.root.name in invalidated:
        assert first_offender is not None  # seeded by step 1 before any propagation
        label, record_name = first_offender
        raise SchemaError(
            f"no valid subschema: removing label {label!r} deletes a mandatory "
            f"field of record {record_name!r}",
            code="algebra.extract-invalidates-root", path=record_name)

    # Step 5: drop invalidated records and any fields (mandatory or not)
    # that still point at one -- an optional field typed to an invalidated
    # record can never be satisfied either, so prune() will remove it, but
    # we drop it here too so the intermediate Schema stays ref-consistent
    # (env values must all be reachable/defined; an invalidated record is
    # about to disappear from the env entirely).
    new_env: Dict[str, Record] = {}
    for name, rec in trimmed.items():
        if name in invalidated:
            continue
        fields = [
            f for f in rec.fields
            if not (isinstance(f.type, Ref) and f.type.name in invalidated)
        ]
        new_env[name] = Record(fields)

    result = Schema(Ref(s.root.name), new_env)
    return normalize(prune(result))

Schema isomorphism -- the paper's Algorithm 3, step 3.

Theorem 4 (the paper): two schemas are equivalent iff their minimized (normalized) forms are isomorphic. That gives a second, algorithm- independent decision procedure for equivalent -- structurally unrelated to bidirectional compatible_with (ops/subschema.py), so the two can be cross-checked against each other in tests (see docs/testing.md, "the dual-algorithm oracle").

_isomorphic stays private to this module -- Schema.equivalent() remains the public API's canonical definition of schema equality, unchanged (issue #279). This module now has two legitimate consumers of the same underlying check: the independent property-test oracle described above, and the public Schema.isomorphic_to() (omnist/schema.py), a deliberately narrower, additional operation for callers who need to detect structural differences that equivalent()'s document-language- only definition cannot see -- e.g. two schemas that accept the same documents only because a bug merged what should have been two distinct records (the exact false-negative equivalent() misses, and _isomorphic catches, that motivated exposing this publicly).

Algorithm: parallel traversal from both roots, building a bijection name_a -> name_b (and its inverse) between env record names as the traversal discovers pairs. At each visited record pair, local_signature must match (same target-blind shape); since local_signature sorts fields by label and ref/scalar shape is part of the key, fields on the two sides line up one-to-one by label once the signatures agree. For each ref-typed field, the two targets are recursively required to be isomorphic, with the bijection enforced consistently in both directions: if a name has already been mapped, revisiting it must reach the same partner every time (and vice versa) -- exactly the DFA-isomorphism check partition refinement's minimal form is built to make trivial: after normalize(), any structural match between two records has to be a consistent renaming, not a coincidence, since minimize has already merged every pair of records that could otherwise masquerade as "the same record under a different name."

Both inputs are assumed already normalized (pruned + minimized) by the caller -- this module does not call normalize itself, matching the paper's Algorithm 3, which runs isomorphism testing as a step after MinimizeSA, not as a self-contained schema comparison.

Field-signature helpers for schema minimization (and, later, isomorphism).

local_signature is the target-blind structural key used as the initial partition for MinimizeSA (issue #140, ops/minimize.py): a key including ref target names would be too strong a starting point -- records that turn out to be equivalent because their ref targets are themselves equivalent-but-differently-named would never even land in the same starting block. It captures a field's label, cardinality, and scalar-or-ref shape, but excludes ref target names (those are compared by evolving block id during refinement instead).

local_signature(rec)

Target-blind structural key for a record: fields sorted by label, each keyed by (label, min, max, shape) where shape is ("scalar", name, nullable) for a scalar field or ("ref",) for a ref field -- the target record's name is deliberately excluded, since minimization must be free to merge records whose ref targets are themselves later found equivalent under different names.

Fields are sorted by label rather than kept in declaration order: validation ignores field order (a Record is a set of labeled fields, per docs/design/model.md), and OSD's printed field order is purely cosmetic. Two records that declare the same fields in a different order accept exactly the same documents and so MUST land in the same initial partition block -- keying by declaration order would incorrectly split them and could prevent them from ever merging.

Source code in omnist/ops/signature.py

def local_signature(
    rec: Record,
) -> tuple[str, tuple[tuple[str, int, int | None, tuple[str, ...] | tuple[str, str, bool]], ...]]:
    """Target-blind structural key for a record: fields sorted by label,
    each keyed by ``(label, min, max, shape)`` where ``shape`` is
    ``("scalar", name, nullable)`` for a scalar field or ``("ref",)`` for a
    ref field -- the target record's *name* is deliberately excluded, since
    minimization must be free to merge records whose ref targets are
    themselves later found equivalent under different names.

    Fields are sorted by label rather than kept in declaration order:
    validation ignores field order (a ``Record`` is a *set* of labeled
    fields, per ``docs/design/model.md``), and OSD's printed field order is
    purely cosmetic. Two records that declare the same fields in a
    different order accept exactly the same documents and so MUST land in
    the same initial partition block -- keying by declaration order would
    incorrectly split them and could prevent them from ever merging.
    """
    fields = tuple(sorted(
        ((f.label, f.min, f.max, _shape_key(f.type)) for f in rec.fields),
        key=lambda t: t[0],
    ))
    return ("record", fields)

---

Adjustment Reports (omnist.report)

Adjustment reports for lossy writes.

Writing a Document to a format that can't hold every value (TOML has no null; JSON/XML have no date type) means the writer has to adjust the data. Each adjustment is recorded as an :class:Adjustment in a :class:WriteReport rather than lost silently. The same report drives three behaviours:

• lenient (default) — adjust and move on; ignore the report if you like.
• inspect — pass report= to a writer (or call check_*) to see what changed without stopping.
• strict — strict=True raises :class:~omnist.errors.WriteError (carrying the report) if anything had to be adjusted.

Each adjustment has a severity: "warning" (conventional / recoverable — a date written as a string) or "error" (likely to surprise or corrupt — a null dropped, NaN in JSON). strict ignores severity and raises on anything.

WriteReport

Everything a writer adjusted. Truthy when there are no error-severity entries (warnings are fine), so if check_toml(doc): … reads as 'safe'.

Source code in omnist/report.py

class WriteReport:
    """Everything a writer adjusted.  Truthy when there are no error-severity
    entries (warnings are fine), so ``if check_toml(doc): …`` reads as 'safe'."""

    def __init__(self) -> None:
        """Initialize an empty WriteReport."""
        self.adjustments: List[Adjustment] = []

    def add(self, path: str, code: str, message: str, severity: str) -> None:
        """Record an adjustment into this report."""
        self.adjustments.append(Adjustment(path, code, message, severity))

    @property
    def warnings(self) -> List[Adjustment]:
        """All adjustments with ``'warning'`` severity."""
        return [a for a in self.adjustments if a.severity == "warning"]

    @property
    def errors(self) -> List[Adjustment]:
        """All adjustments with ``'error'`` severity."""
        return [a for a in self.adjustments if a.severity == "error"]

    def __bool__(self) -> bool:
        return not self.errors

    def __iter__(self) -> Iterator[Adjustment]:
        return iter(self.adjustments)

    def __len__(self) -> int:
        return len(self.adjustments)

    def __str__(self) -> str:
        if not self.adjustments:
            return "no adjustments"
        return "\n".join(f"{a.severity}: {a.path}: {a.message}" for a in self.adjustments)

errors property

All adjustments with 'error' severity.

warnings property

All adjustments with 'warning' severity.

__init__()

Initialize an empty WriteReport.

Source code in omnist/report.py

def __init__(self) -> None:
    """Initialize an empty WriteReport."""
    self.adjustments: List[Adjustment] = []

add(path, code, message, severity)

Record an adjustment into this report.

Source code in omnist/report.py

def add(self, path: str, code: str, message: str, severity: str) -> None:
    """Record an adjustment into this report."""
    self.adjustments.append(Adjustment(path, code, message, severity))

finish_write(text, rep, *, strict=False, report=None)

Apply the standard strict / report handling to a writer's result.

If report is given, rep's adjustments are copied into it. If strict and rep has any adjustments, raises WriteError carrying rep. Otherwise returns text.

Source code in omnist/report.py

def finish_write(text: str, rep: WriteReport, *, strict: bool = False,
                 report: Optional[WriteReport] = None) -> str:
    """Apply the standard ``strict`` / ``report`` handling to a writer's result.

    If ``report`` is given, ``rep``'s adjustments are copied into it.  If
    ``strict`` and ``rep`` has any adjustments, raises ``WriteError`` carrying
    ``rep``.  Otherwise returns ``text``.
    """
    if report is not None:
        report.adjustments.extend(rep.adjustments)
    if strict and rep.adjustments:
        raise WriteError(str(rep), report=rep)
    return text

---

Format Registry (omnist.registry)

Format registry — read/write a Document by format name, and register plugins.

A :class:Format bundles a name with read(text) -> node and write(node, **opts) -> str callables, and an optional check(node) -> WriteReport for simulating a write without producing output (Doc.check_format needs it; read/write alone don't). The four built-ins register themselves on import; :func:register_format adds your own, usable everywhere (including Doc.from_format / Doc.to_format).

formats()

The names of all registered formats, sorted.

Source code in omnist/registry.py

def formats() -> List[str]:
    """The names of all registered formats, sorted."""
    with _LOCK:
        return sorted(_REGISTRY)

get_format(name)

The registered :class:Format for name (raises if unknown).

Source code in omnist/registry.py

def get_format(name: str) -> Format:
    """The registered :class:`Format` for ``name`` (raises if unknown)."""
    with _LOCK:
        try:
            return _REGISTRY[name]
        except KeyError:
            known = ", ".join(sorted(_REGISTRY)) or "(none)"
            raise OmnistError(f"unknown format {name!r}; registered: {known}") from None

register_format(fmt)

Register (or replace) a format plugin.

Source code in omnist/registry.py

def register_format(fmt: Format) -> None:
    """Register (or replace) a format plugin."""
    with _LOCK:
        _REGISTRY[fmt.name] = fmt

---

Errors & Exceptions (omnist.errors)

Exceptions (and one warning) used across omnist.

DetachedNode

Bases: DocumentError

A cursor was used after its node was removed from the document.

Holding a :class:~omnist.document.Doc cursor and then removing that node (or a node above it) leaves the cursor pointing at a subtree no longer in the document. Using it raises this instead of silently editing an orphan.

Source code in omnist/errors.py

class DetachedNode(DocumentError):
    """A cursor was used after its node was removed from the document.

    Holding a :class:`~omnist.document.Doc` cursor and then removing that node
    (or a node above it) leaves the cursor pointing at a subtree no longer in the
    document.  Using it raises this instead of silently editing an orphan.
    """

DocumentError

Bases: OmnistError

A Python value is not a legal Document, or a Document operation is invalid.

Raised by the :class:~omnist.document.Doc API when an import or mutation would produce something outside the Document model — an unsupported Python type, a non-string object key, a cycle — or when an operation doesn't fit the node (e.g. get on a scalar). The message carries the offending path.

code/path are optional structured attributes, None unless the raiser passed them. The reader-side failures that have a document.* code (docs/08-conformance-and-errors.md Sec8.3.2: a safety limit exceeded, an input construct with no label to become an edge) set them; path is then a Document path (E-11), never a text position.

Source code in omnist/errors.py

class DocumentError(OmnistError):
    """A Python value is not a legal Document, or a Document operation is invalid.

    Raised by the :class:`~omnist.document.Doc` API when an import or mutation
    would produce something outside the Document model — an unsupported Python
    type, a non-string object key, a cycle — or when an operation doesn't fit the
    node (e.g. ``get`` on a scalar).  The message carries the offending path.

    ``code``/``path`` are optional structured attributes, ``None`` unless the
    raiser passed them. The reader-side failures that have a ``document.*``
    code (``docs/08-conformance-and-errors.md`` Sec8.3.2: a safety limit
    exceeded, an input construct with no label to become an edge) set them;
    ``path`` is then a Document path (E-11), never a text position.
    """

    def __init__(self, message: str, *, code: "Optional[str]" = None,
                 path: "Optional[str]" = None) -> None:
        """Initialize DocumentError with a human-readable message and,
        optionally, a structured code and the Document path it applies to."""
        super().__init__(message)
        self.code = code
        self.path = path

__init__(message, *, code=None, path=None)

Initialize DocumentError with a human-readable message and, optionally, a structured code and the Document path it applies to.

Source code in omnist/errors.py

def __init__(self, message: str, *, code: "Optional[str]" = None,
             path: "Optional[str]" = None) -> None:
    """Initialize DocumentError with a human-readable message and,
    optionally, a structured code and the Document path it applies to."""
    super().__init__(message)
    self.code = code
    self.path = path

OmnistError

Bases: Exception

Base class for all omnist errors.

Source code in omnist/errors.py

class OmnistError(Exception):
    """Base class for all omnist errors."""

ParseError

Bases: OmnistError

A document could not be read from its format (outside the supported profile).

Format-syntax failures (invalid JSON/YAML/TOML/XML/OML text) carry code/path (issue #308) -- optional structured attributes, None unless the raiser passed them, so every existing raise ParseError("msg") call keeps working unchanged -- but .errors stays empty, the same way :class:SchemaError distinguishes a single lexical/well-formedness problem from a collected list: a syntax failure stops parsing at the first error, so there is nothing to collect. Schema-conformance failures from :func:~omnist.deserialize.materialize go the other way: they carry the full structured .errors list of every problem found (path, message, machine-readable code), not just the first one, so callers -- an API server turning this into a JSON error response, for instance -- can inspect and report on each one individually instead of parsing str(exc); code/path stay unset for this case, since there's no single position to point at.

Source code in omnist/errors.py

class ParseError(OmnistError):
    """A document could not be read from its format (outside the supported profile).

    Format-syntax failures (invalid JSON/YAML/TOML/XML/OML text) carry
    ``code``/``path`` (issue #308) -- optional structured attributes, ``None``
    unless the raiser passed them, so every existing ``raise
    ParseError("msg")`` call keeps working unchanged -- but ``.errors`` stays
    empty, the same way :class:`SchemaError` distinguishes a single
    lexical/well-formedness problem from a collected list: a syntax failure
    stops parsing at the first error, so there is nothing to collect.
    Schema-conformance failures from :func:`~omnist.deserialize.materialize`
    go the other way: they carry the full structured ``.errors`` list of
    every problem found (path, message, machine-readable code), not just the
    first one, so callers -- an API server turning this into a JSON error
    response, for instance -- can inspect and report on each one
    individually instead of parsing ``str(exc)``; ``code``/``path`` stay
    unset for this case, since there's no single position to point at.
    """

    def __init__(self, message: str, errors: "Optional[List[Error]]" = None, *,
                 code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
        """Initialize ParseError with a human-readable message and either
        structured per-problem issues (materialize) or a structured
        code/path for a single syntax failure -- never both at once."""
        super().__init__(message)
        self.errors: "List[Error]" = errors or []
        self.code = code
        self.path = path

__init__(message, errors=None, *, code=None, path=None)

Initialize ParseError with a human-readable message and either structured per-problem issues (materialize) or a structured code/path for a single syntax failure -- never both at once.

Source code in omnist/errors.py

def __init__(self, message: str, errors: "Optional[List[Error]]" = None, *,
             code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
    """Initialize ParseError with a human-readable message and either
    structured per-problem issues (materialize) or a structured
    code/path for a single syntax failure -- never both at once."""
    super().__init__(message)
    self.errors: "List[Error]" = errors or []
    self.code = code
    self.path = path

SchemaError

Bases: OmnistError

The schema text or structure is invalid.

code/path are optional structured attributes -- None unless the raiser passed them, so every existing raise SchemaError("msg") call keeps working unchanged. Where set, code is one of omnist-spec's parse.*/schema.* taxonomy codes (see docs/08-conformance-and-errors.md Sec8.3.1/8.3.3 in the omnist-spec submodule) and path is the OSD text offset or record/field name the problem was found at. Unlike :class:ParseError, a single SchemaError always represents exactly one problem -- OSD parsing stops at the first error, so there is no .errors list to collect (issue #301).

Source code in omnist/errors.py

class SchemaError(OmnistError):
    """The schema text or structure is invalid.

    ``code``/``path`` are optional structured attributes -- ``None`` unless
    the raiser passed them, so every existing ``raise SchemaError("msg")``
    call keeps working unchanged. Where set, ``code`` is one of
    ``omnist-spec``'s ``parse.*``/``schema.*`` taxonomy codes (see
    ``docs/08-conformance-and-errors.md`` Sec8.3.1/8.3.3 in the
    ``omnist-spec`` submodule) and ``path`` is the OSD text offset or
    record/field name the problem was found at. Unlike :class:`ParseError`,
    a single ``SchemaError`` always represents exactly one problem -- OSD
    parsing stops at the first error, so there is no ``.errors`` list to
    collect (issue #301).
    """

    def __init__(self, message: str, *, code: "Optional[str]" = None,
                 path: "Optional[str]" = None) -> None:
        """Initialize SchemaError with a human-readable message and, optionally,
        a structured machine-readable code and the path/position it applies to."""
        super().__init__(message)
        self.code = code
        self.path = path

__init__(message, *, code=None, path=None)

Initialize SchemaError with a human-readable message and, optionally, a structured machine-readable code and the path/position it applies to.

Source code in omnist/errors.py

def __init__(self, message: str, *, code: "Optional[str]" = None,
             path: "Optional[str]" = None) -> None:
    """Initialize SchemaError with a human-readable message and, optionally,
    a structured machine-readable code and the path/position it applies to."""
    super().__init__(message)
    self.code = code
    self.path = path

UnsafeXMLWarning

Bases: UserWarning

Unused by read_xml as of the fix for the fail-open XML fallback (see issue #173) — defusedxml is now a hard requirement for XML support, and its absence raises ImportError instead of falling back to the unsafe standard-library parser with a warning. Kept exported for backward compatibility with any code that imports or references it (e.g. an existing warnings.filterwarnings(..., category=omnist.UnsafeXMLWarning) call), but nothing in omnist raises it anymore.

Source code in omnist/errors.py

class UnsafeXMLWarning(UserWarning):
    """Unused by ``read_xml`` as of the fix for the fail-open XML fallback
    (see issue #173) — ``defusedxml`` is now a hard requirement for XML
    support, and its absence raises ``ImportError`` instead of falling back
    to the unsafe standard-library parser with a warning. Kept exported for
    backward compatibility with any code that imports or references it
    (e.g. an existing ``warnings.filterwarnings(..., category=omnist.UnsafeXMLWarning)``
    call), but nothing in omnist raises it anymore.
    """

WriteError

Bases: OmnistError

A document cannot be represented in the target format.

Raised in strict=True mode for any recorded adjustment, and unconditionally (regardless of strict) when the value has no legal representation at all in the target format -- see docs/08-conformance-and-errors.md Sec8.3.8/8.3.9 in the omnist-spec submodule -- carrying code="write.unsupported-value" and the offending path (issues #323/#324/#325). code/path are optional structured attributes, None unless the raiser passed them, so every existing raise WriteError("msg") call keeps working unchanged. .report holds the full :class:~omnist.report.WriteReport of every adjustment that would have been needed (empty for an unconditional failure raised before any adjustment was recorded), so callers can inspect the structured list, not just the text.

Source code in omnist/errors.py

class WriteError(OmnistError):
    """A document cannot be represented in the target format.

    Raised in ``strict=True`` mode for any recorded adjustment, and
    unconditionally (regardless of ``strict``) when the value has no legal
    representation at all in the target format -- see
    ``docs/08-conformance-and-errors.md`` Sec8.3.8/8.3.9 in the
    ``omnist-spec`` submodule -- carrying ``code="write.unsupported-value"``
    and the offending ``path`` (issues #323/#324/#325). ``code``/``path`` are
    optional structured attributes, ``None`` unless the raiser passed them,
    so every existing ``raise WriteError("msg")`` call keeps working
    unchanged. ``.report`` holds the full
    :class:`~omnist.report.WriteReport` of every adjustment that would have
    been needed (empty for an unconditional failure raised before any
    adjustment was recorded), so callers can inspect the structured list,
    not just the text.
    """

    def __init__(self, message: str, report: "WriteReport | None" = None, *,
                 code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
        """Initialize WriteError with an adjustment report and, optionally,
        a structured machine-readable code and the path it applies to."""
        super().__init__(message)
        self.report = report
        self.code = code
        self.path = path

__init__(message, report=None, *, code=None, path=None)

Initialize WriteError with an adjustment report and, optionally, a structured machine-readable code and the path it applies to.

Source code in omnist/errors.py

def __init__(self, message: str, report: "WriteReport | None" = None, *,
             code: "Optional[str]" = None, path: "Optional[str]" = None) -> None:
    """Initialize WriteError with an adjustment report and, optionally,
    a structured machine-readable code and the path it applies to."""
    super().__init__(message)
    self.report = report
    self.code = code
    self.path = path

---

CLI (omnist.cli)

The omnist command-line interface.

A thin wrapper over the public :mod:omnist API -- see docs/design/cli-spec.md for the full command surface. Each command maps to one or two calls into the library; this module adds no new behavior of its own beyond argument parsing, file/stdio plumbing, and exit codes.

main(argv=None)

Entry point for the omnist command-line interface.

Source code in omnist/cli.py

def main(argv: Optional[Sequence[str]] = None) -> int:
    """Entry point for the ``omnist`` command-line interface."""
    parser = _build_parser()
    args = parser.parse_args(argv)
    try:
        return args.func(args)  # type: ignore[no-any-return]
    except (ParseError, SchemaError, WriteError, DocumentError, OSError) as exc:
        if getattr(args, "json", False):
            print(_json_error(exc))
        else:
            print(f"error: {exc}", file=sys.stderr)
        return 2