Schema inspection

bondi.schema.Schema dataclass

Infer an observed schema from dictionaries without retaining source documents.

Parameters:

Name	Type	Description	Default
values	ValueMode	Primitive retention: none (default), distinct values, or full exact frequencies. The policy remains fixed across updates. Both enabled modes use memory proportional to distinct values and impose no memory limit.	'none'

Objects use dict (including subclasses), arrays use list. Scalar support includes JSON primitives, bytes, datetime/date, finite Decimal, UUID and BSON ObjectId, Int64 and Binary. Other scalar types and nonfinite numbers raise explicitly. Nesting beyond 128 levels is rejected before accumulation. No JSON text decoding, database sampling or backend query semantics are included.

Source code in src/bondi/schema.py

@dataclass(init=False)
class Schema:
    """Infer an observed schema from dictionaries without retaining source documents.

    Args:
        values: Primitive retention: none (default), distinct values, or full exact
            frequencies. The policy remains fixed across updates. Both enabled modes
            use memory proportional to distinct values and impose no memory limit.

    Objects use dict (including subclasses), arrays use list. Scalar support includes
    JSON primitives, bytes, datetime/date, finite Decimal, UUID and BSON ObjectId,
    Int64 and Binary. Other scalar types and nonfinite numbers raise explicitly.
    Nesting beyond 128 levels is rejected before accumulation. No JSON text decoding,
    database sampling or backend query semantics are included.
    """

    _mode: ValueMode
    _count: int
    _fields: dict[str, FieldSchema] = field(repr=False)
    _flat: dict[Path, FieldSchema] = field(repr=False)

    def __init__(self, *, values: ValueMode = "none") -> None:
        if values not in ("none", "distinct", "full"):
            raise ValueError("values must be 'none', 'distinct' or 'full'")
        self._mode = values
        self._count = 0
        self._fields = {}
        self._flat = {}

    def __repr__(self) -> str:
        return (
            f"Schema(document_count={self.document_count}, paths={len(self._flat)}, "
            f"values={self.value_mode!r})"
        )

    @classmethod
    def from_documents(cls, documents: Iterable[dict], *, values: ValueMode = "none") -> "Schema":
        """Construct a schema from one pass over an iterable of documents.

        Accepts generators and empty iterables. Each document is validated before
        accumulation; input dictionaries and lists are not modified or retained.
        """
        schema = cls(values=values)
        schema.update(documents)
        return schema

    @property
    def document_count(self) -> int:
        """Number of successfully ingested root documents, including empty objects."""
        return self._count

    @property
    def value_mode(self) -> ValueMode:
        """Fixed primitive-value retention policy."""
        return self._mode

    @property
    def fields(self) -> Mapping[str, FieldSchema]:
        """Live top-level fields keyed by literal field names."""
        return MappingProxyType(self._fields)

    @property
    def flat_fields(self) -> Mapping[str, FieldSchema]:
        """All locations keyed by escaped display paths, including array elements.

        Keys are a snapshot; values reference live field objects. Backslash escapes
        literal dots, opening brackets and backslashes. Tuple lookup avoids escaping.
        """
        return MappingProxyType({_display(path): value for path, value in self._flat.items()})

    @property
    def primitive_fields(self) -> Mapping[str, FieldSchema]:
        """Locations with any primitive type, even if objects/arrays also occur."""
        return MappingProxyType(
            {
                _display(path): value
                for path, value in self._flat.items()
                if any(name not in ("object", "array") for name in value.types)
            }
        )

    def __getitem__(self, path: str | Path) -> FieldSchema:
        """Look up an exact path; raise KeyError if absent. Arrays require explicit []."""
        return self._flat[_path(path)]

    def update(self, documents: Iterable[dict]) -> None:
        """Add a batch, keeping earlier documents if a later one fails validation.

        TypeError reports unsupported values or keys; ValueError reports cycles or
        nonfinite values. A failing document contributes nothing. Iterator failures
        also preserve earlier documents. Calls are synchronous and not thread-safe.
        """
        for document in documents:
            if not isinstance(document, dict):
                raise TypeError("Each root document must be a dictionary")
            _validate(document, set())
            self._count += 1
            self._object(document, self, ())

    def _object(self, document, parent, prefix):
        for name, value in document.items():
            path = prefix + (name,)
            if name not in parent._fields:
                parent._fields[name] = FieldSchema(path, _parent=parent)
                self._flat[path] = parent._fields[name]
            self._observe(parent._fields[name], value)

    def _observe(self, node, value):
        name = _type_name(value)
        node.count += 1
        if name not in node._types:
            variant = TypeSchema(name, _mode=self._mode)
            node._types[name] = variant
            if name == "array":
                path = node.path + (ARRAY_ELEMENT,)
                variant.elements = FieldSchema(path)
                self._flat[path] = variant.elements
            elif name != "object" and self._mode != "none":
                variant._retained = Counter() if self._mode == "full" else set()
        variant = node._types[name]
        variant.count += 1
        if name == "object":
            self._object(value, variant, node.path)
        elif name == "array":
            for item in value:
                self._observe(variant.elements, item)
        elif self._mode == "full":
            variant._retained[value] += 1
        elif self._mode == "distinct":
            variant._retained.add(value)

document_count property

Number of successfully ingested root documents, including empty objects.

fields property

Live top-level fields keyed by literal field names.

flat_fields property

All locations keyed by escaped display paths, including array elements.

Keys are a snapshot; values reference live field objects. Backslash escapes literal dots, opening brackets and backslashes. Tuple lookup avoids escaping.

primitive_fields property

Locations with any primitive type, even if objects/arrays also occur.

value_mode property

Fixed primitive-value retention policy.

__getitem__(path)

Look up an exact path; raise KeyError if absent. Arrays require explicit [].

Source code in src/bondi/schema.py

def __getitem__(self, path: str | Path) -> FieldSchema:
    """Look up an exact path; raise KeyError if absent. Arrays require explicit []."""
    return self._flat[_path(path)]

from_documents(documents, *, values='none') classmethod

Construct a schema from one pass over an iterable of documents.

Accepts generators and empty iterables. Each document is validated before accumulation; input dictionaries and lists are not modified or retained.

Source code in src/bondi/schema.py

@classmethod
def from_documents(cls, documents: Iterable[dict], *, values: ValueMode = "none") -> "Schema":
    """Construct a schema from one pass over an iterable of documents.

    Accepts generators and empty iterables. Each document is validated before
    accumulation; input dictionaries and lists are not modified or retained.
    """
    schema = cls(values=values)
    schema.update(documents)
    return schema

update(documents)

Add a batch, keeping earlier documents if a later one fails validation.

TypeError reports unsupported values or keys; ValueError reports cycles or nonfinite values. A failing document contributes nothing. Iterator failures also preserve earlier documents. Calls are synchronous and not thread-safe.

Source code in src/bondi/schema.py

def update(self, documents: Iterable[dict]) -> None:
    """Add a batch, keeping earlier documents if a later one fails validation.

    TypeError reports unsupported values or keys; ValueError reports cycles or
    nonfinite values. A failing document contributes nothing. Iterator failures
    also preserve earlier documents. Calls are synchronous and not thread-safe.
    """
    for document in documents:
        if not isinstance(document, dict):
            raise TypeError("Each root document must be a dictionary")
        _validate(document, set())
        self._count += 1
        self._object(document, self, ())

bondi.schema.FieldSchema dataclass

Observations at an object key or array-element location.

Attributes:

Name	Type	Description
path	Path	Unambiguous structural path: field-name strings and ARRAY_ELEMENT.
count	int	Number of times this field or array element was observed.

parent_count/missing_count refer to immediate parent objects, not root-document coverage. They are None for array elements. Schema owns these instances; references remain live across update() calls.

Source code in src/bondi/schema.py

@dataclass
class FieldSchema:
    """Observations at an object key or array-element location.

    Attributes:
        path: Unambiguous structural path: field-name strings and ARRAY_ELEMENT.
        count: Number of times this field or array element was observed.

    parent_count/missing_count refer to immediate parent objects, not root-document
    coverage. They are None for array elements. Schema owns these instances;
    references remain live across update() calls.
    """

    path: Path
    count: int = 0
    _parent: "Schema | TypeSchema | None" = field(default=None, repr=False, compare=False)
    _types: dict[str, TypeSchema] = field(default_factory=dict, repr=False)

    @property
    def types(self) -> Mapping[str, TypeSchema]:
        """Observed types in first-encounter order, exposed as a read-only mapping."""
        return MappingProxyType(self._types)

    @property
    def is_multi_type(self) -> bool:
        """Whether more than one type has been observed at this location."""
        return len(self._types) > 1

    @property
    def parent_count(self) -> int | None:
        """Number of parent objects; unavailable for array-element locations."""
        if self._parent is None:
            return None
        if isinstance(self._parent, Schema):
            return self._parent.document_count
        return self._parent.count

    @property
    def missing_count(self) -> int | None:
        """Parent objects lacking this key; null values count as present."""
        parent = self.parent_count
        return None if parent is None else parent - self.count

    @property
    def has_missing(self) -> bool | None:
        """Whether any parent objects lack this key; None for array elements."""
        missing = self.missing_count
        return None if missing is None else missing > 0

has_missing property

Whether any parent objects lack this key; None for array elements.

is_multi_type property

Whether more than one type has been observed at this location.

missing_count property

Parent objects lacking this key; null values count as present.

parent_count property

Number of parent objects; unavailable for array-element locations.

types property

Observed types in first-encounter order, exposed as a read-only mapping.

bondi.schema.TypeSchema dataclass

Observations of one type at one field or array-element location.

Attributes:

Name	Type	Description
name	str	Type label: object, array, null, or the supported scalar's Python name.
count	int	Number of occurrences of this type, including repeated array values.
elements	FieldSchema | None	Array-element schema for arrays; otherwise None. Empty arrays still have an element schema with count zero and no observed types.

Instances are maintained by Schema; inspect them rather than modifying counters. Retained values are immutable and separated by type. Container types never retain complete objects/arrays as values. Fields and count mappings are read-only views.

Source code in src/bondi/schema.py

@dataclass
class TypeSchema:
    """Observations of one type at one field or array-element location.

    Attributes:
        name: Type label: object, array, null, or the supported scalar's Python name.
        count: Number of occurrences of this type, including repeated array values.
        elements: Array-element schema for arrays; otherwise None. Empty arrays
            still have an element schema with count zero and no observed types.

    Instances are maintained by Schema; inspect them rather than modifying counters.
    Retained values are immutable and separated by type. Container types never retain
    complete objects/arrays as values. Fields and count mappings are read-only views.
    """

    name: str
    count: int = 0
    elements: "FieldSchema | None" = None
    _mode: ValueMode = field(default="none", repr=False)
    _fields: dict[str, "FieldSchema"] = field(default_factory=dict, repr=False)
    _retained: set | Counter | None = field(default=None, repr=False)

    @property
    def fields(self) -> Mapping[str, "FieldSchema"]:
        """Object children, keyed by literal field name; empty for other types."""
        return MappingProxyType(self._fields)

    @property
    def values(self) -> frozenset | None:
        """Distinct retained primitives; None if disabled or this is a container.

        In full mode this is derived from the counter, not a second stored set.
        The returned frozenset is a snapshot; request it again after an update.
        """
        return None if self._retained is None else frozenset(self._retained)

    @property
    def value_counts(self) -> Mapping[Any, int] | None:
        """Live read-only exact frequencies in full mode; otherwise None."""
        if self._mode != "full" or self._retained is None:
            return None
        return MappingProxyType(self._retained)

fields property

Object children, keyed by literal field name; empty for other types.

value_counts property

Live read-only exact frequencies in full mode; otherwise None.

values property

Distinct retained primitives; None if disabled or this is a container.

In full mode this is derived from the counter, not a second stored set. The returned frozenset is a snapshot; request it again after an update.

Array paths and value retention

ARRAY_ELEMENT identifies an array-element step in tuple paths, for example schema[("tags", ARRAY_ELEMENT)]. Import it from bondi.schema. ValueMode is the literal type "none" | "distinct" | "full".

Exact paths

from bondi.schema import ARRAY_ELEMENT

schema["items.[].price"]
schema[("items", ARRAY_ELEMENT, "price")]  # Same lookup, without string syntax.

Tuple paths distinguish literal keys from structural separators. For a document containing both {"a.b": 1} and {"a": {"b": 2}}:

schema[("a.b",)]  # Literal dotted key.
schema[("a", "b")]  # Nested key; equivalent to schema["a.b"].

flat_fields uses escaped display strings: backslash escapes dots, opening brackets and backslashes in literal keys. Every displayed key can be passed back to schema[path]. For example, r"a\.b" selects the literal dotted key; r"\[]" selects a literal key named []. Empty-string keys remain supported. Unknown paths raise KeyError; no implicit MongoDB-style array traversal occurs.

Incremental ingestion and ownership

schema = Schema(values="distinct")
schema.update(first_batch)
city = schema["city"]
schema.update(second_batch)
# city now reflects both batches; schema.value_mode remains "distinct".

Generators are accepted and source containers are not retained or modified. Each document is validated before accumulation. If a later document or the input iterator fails, earlier documents remain, but an invalid document contributes no counts. Nesting beyond 128 levels, cycles, non-string keys and unsupported values are rejected. Calls are synchronous and not thread-safe; do not modify input during ingestion.

Field/type objects remain live after updates. Treat them as inspection objects and do not assign their counters or internals. Mapping properties are read-only; fields, types and value_counts are live views. flat_fields and primitive_fields have snapshot keys referencing live field objects; request them again to discover newly added paths. .values is an immutable snapshot.

Supported data and database integration

Objects are dictionaries (including subclasses), arrays are lists. Supported scalars are str, bool, int, finite float, None, bytes, datetime, date, finite Decimal, UUID, and BSON ObjectId, Int64, Binary. Type names follow Python class names except object, array, and null. Unsupported BSON types such as Decimal128 currently raise TypeError; they are not silently normalized.

Pass already-decoded documents. JSON text parsing and conversion policies belong to the caller. The following queries use the shared demo data from our local Docker setup. Use a MongoDB executor for the first two and a Couchbase executor for the others; see the MongoDB and Couchbase connection examples.

MongoDB: a bounded preview

from bondi.mongodb import MongoCollectionRef, MongoFindQuery

sample_query = MongoFindQuery(
    MongoCollectionRef("demo", "business"),
    filter={},
    limit=100,
)

This returns up to 100 whole documents. A limit alone is not a random sample; without an explicit sort, do not assume a stable selection or representative coverage.

MongoDB: random sampling

from bondi.mongodb import MongoAggregateQuery, MongoCollectionRef

sample_query = MongoAggregateQuery(
    MongoCollectionRef("demo", "business"),
    pipeline=[{"$sample": {"size": 100}}],
)

MongoDB's $sample selects documents randomly. With $sample first, a collection larger than 100 documents and a requested sample smaller than 5% of the collection, MongoDB can use a pseudo-random cursor; otherwise sampling can require scanning and a random top-k sort. The tiny demo collection contains only eight documents, so it cannot supply 100 distinct source documents. Repeated sampling calls need not return the same documents.

Couchbase: a bounded preview

from bondi.couchbase import CouchbaseQuery

sample_query = CouchbaseQuery(
    "SELECT RAW b FROM `demo`.`_default`.`business` AS b LIMIT 100"
)

SELECT RAW b returns whole document bodies rather than { "b": ... } alias wrappers. This matters because the parser would otherwise infer an extra field named b. As with MongoDB's limit, this is a bounded preview, not random sampling.

Couchbase: random selection through SQL++

For the current SQL++ executor, random ordering followed by a limit is an option:

sample_query = CouchbaseQuery(
    "SELECT RAW b FROM `demo`.`_default`.`business` AS b "
    "ORDER BY RANDOM() LIMIT 100"
)

RANDOM() produces pseudo-random numbers. This query is useful for small datasets but may scan and randomly sort a large candidate set; LIMIT bounds the output, not the amount of database work. It is not an efficient sampling primitive equivalent to MongoDB's random-cursor path. This query was checked against both local 7.6.2 editions.

Couchbase also offers a dedicated SamplingScan with a limit and optional seed, via the SDK's key-value collection scan API. It is not a SQL++ query operator and is not currently exposed by AsyncCouchbaseExecutor. Supporting that would be a separate sampling feature; its suitability and statistical properties should be evaluated before adopting it.

Infer the schema from any of these queries

Inside the corresponding executor context or notebook:

from bondi.schema import Schema

result = executor.run(sample_query, collect_results=True)
schema = Schema.from_documents(result.documents)
print(schema.document_count)
print(list(schema.primitive_fields))

Retrieve whole document bodies: projections change the schema you observe. Save sampled documents when an experiment must reuse exactly the same input. No database sampling helper, schema persistence, merging, or validation-schema export is introduced here.

Run notebooks/schema_demo.ipynb to explore all modes using a tiny in-memory dataset; no database or Couchbase SDK is needed.