MongoDB queries

`bondi.mongodb.MongoCollectionRef` `dataclass`

Identify a MongoDB collection independently of a deployment.

References compare equal when both names match. They are immutable and hashable, so they can be dictionary keys or set members. No connection is created and collection existence is not checked.

Names are preserved exactly, including case and whitespace. Validation catches empty names and null characters; it does not implement all MongoDB server naming restrictions or guarantee that a server accepts the names.

Parameters:

Name	Type	Description	Default
`database`	`str`	Nonempty database name without null characters.	required
`collection`	`str`	Nonempty collection name without null characters.	required

Raises:

Type	Description
`TypeError`	Either name is not a string.
`ValueError`	Either name is empty or contains a null character.

Examples:

>>> target = MongoCollectionRef("research", "businesses")
>>> target.database
'research'
>>> target == MongoCollectionRef("research", "businesses")
True
>>> len({target, MongoCollectionRef("research", "businesses")})
1

Source code in src/bondi/mongodb/collection.py

@dataclass(frozen=True)
class MongoCollectionRef:
    """Identify a MongoDB collection independently of a deployment.

    References compare equal when both names match. They are immutable and
    hashable, so they can be dictionary keys or set members. No connection is
    created and collection existence is not checked.

    Names are preserved exactly, including case and whitespace. Validation
    catches empty names and null characters; it does not implement all MongoDB
    server naming restrictions or guarantee that a server accepts the names.

    Args:
        database: Nonempty database name without null characters.
        collection: Nonempty collection name without null characters.

    Raises:
        TypeError: Either name is not a string.
        ValueError: Either name is empty or contains a null character.

    Examples:
        >>> target = MongoCollectionRef("research", "businesses")
        >>> target.database
        'research'
        >>> target == MongoCollectionRef("research", "businesses")
        True
        >>> len({target, MongoCollectionRef("research", "businesses")})
        1
    """

    database: str
    collection: str

    def __post_init__(self) -> None:
        for field_name in ("database", "collection"):
            value = getattr(self, field_name)
            if not isinstance(value, str):
                raise TypeError(f"{field_name} must be a string")
            if not value:
                raise ValueError(f"{field_name} must not be empty")
            if "\0" in value:
                raise ValueError(f"{field_name} must not contain null characters")

`bondi.mongodb.MongoFindQuery` `dataclass`

Describe a MongoDB find operation without executing it.

This is a mutable, unhashable value object. Filter and projection documents are deep-copied on construction, including nested dictionaries and lists. Subsequent edits to those input documents do not affect this query, and vice versa. Values must support Python's deepcopy operation.

Validation runs at construction and before serialization or fingerprinting. Later assignments and nested edits are ordinary Python mutations: they are neither validated nor defensively copied. Validation checks the outer document types and pagination/sort structure; it does not validate MongoDB operators, BSON encodability or server acceptance.

Parameters:

Name	Type	Description	Default
`target`	`MongoCollectionRef`	Logical collection to query, independent of a deployment.	required
`filter`	`dict[str, Any]`	Required predicate dictionary; use an empty dictionary explicitly to describe an unrestricted query.	required
`projection`	`dict[str, Any] \| None`	Optional projection dictionary. None means no explicit projection.	`None`
`sort`	`tuple[tuple[str, int], ...]`	Ordered tuple of (field name, direction) tuples. Directions must be integer 1 or -1. The empty tuple means no explicit sort. Metadata sorts are not supported by this initial interface.	`()`
`skip`	`int`	Nonnegative integer number of results to skip. Booleans are rejected.	`0`
`limit`	`int \| None`	Positive integer result limit, or None for no limit. Zero and booleans are rejected; use None explicitly for an unlimited query.	`None`

Raises:

Type	Description
`TypeError`	A field has an incorrect type, including a boolean used as an integer.
`ValueError`	Pagination bounds or sort names, pair lengths or directions are invalid.

Examples:

>>> predicate = {"stars": {"$gte": 4}}
>>> query = MongoFindQuery(MongoCollectionRef("research", "businesses"), predicate)
>>> query.filter["stars"]["$gte"] = 5
>>> predicate["stars"]["$gte"]
4
>>> query.limit = 20
>>> query.limit
20

Source code in src/bondi/mongodb/query.py

@dataclass
class MongoFindQuery:
    """Describe a MongoDB find operation without executing it.

    This is a mutable, unhashable value object. Filter and projection documents
    are deep-copied on construction, including nested dictionaries and lists.
    Subsequent edits to those input documents do not affect this query, and vice
    versa. Values must support Python's ``deepcopy`` operation.

    Validation runs at construction and before serialization or fingerprinting.
    Later assignments and nested edits are
    ordinary Python mutations: they are neither validated nor defensively copied.
    Validation checks the outer document types and pagination/sort structure;
    it does not validate MongoDB operators, BSON encodability or server acceptance.

    Args:
        target: Logical collection to query, independent of a deployment.
        filter: Required predicate dictionary; use an empty dictionary explicitly
            to describe an unrestricted query.
        projection: Optional projection dictionary. None means no explicit projection.
        sort: Ordered tuple of (field name, direction) tuples. Directions must be
            integer 1 or -1. The empty tuple means no explicit sort. Metadata sorts
            are not supported by this initial interface.
        skip: Nonnegative integer number of results to skip. Booleans are rejected.
        limit: Positive integer result limit, or None for no limit. Zero and booleans
            are rejected; use None explicitly for an unlimited query.

    Raises:
        TypeError: A field has an incorrect type, including a boolean used as an integer.
        ValueError: Pagination bounds or sort names, pair lengths or directions are invalid.

    Examples:
        >>> predicate = {"stars": {"$gte": 4}}
        >>> query = MongoFindQuery(MongoCollectionRef("research", "businesses"), predicate)
        >>> query.filter["stars"]["$gte"] = 5
        >>> predicate["stars"]["$gte"]
        4
        >>> query.limit = 20
        >>> query.limit
        20
    """

    backend: ClassVar[str] = "mongodb"

    target: MongoCollectionRef
    filter: dict[str, Any]
    projection: dict[str, Any] | None = None
    sort: tuple[tuple[str, int], ...] = ()
    skip: int = 0
    limit: int | None = None

    def __post_init__(self) -> None:
        self.validate()
        self.filter = deepcopy(self.filter)
        self.projection = deepcopy(self.projection)

    def validate(self) -> None:
        """Check current structural constraints without modifying the query.

        Raises:
            TypeError: A field has an incorrect type.
            ValueError: A field violates the constructor's value constraints.

        BSON encoding and MongoDB operator validity are not checked here.
        """
        if not isinstance(self.target, MongoCollectionRef):
            raise TypeError("target must be a MongoCollectionRef")
        if not isinstance(self.filter, dict):
            raise TypeError("filter must be a dictionary")
        if self.projection is not None and not isinstance(self.projection, dict):
            raise TypeError("projection must be a dictionary or None")

        if type(self.skip) is not int:
            raise TypeError("skip must be an integer, not a boolean")
        if self.skip < 0:
            raise ValueError("skip must be nonnegative")
        if self.limit is not None:
            if type(self.limit) is not int:
                raise TypeError("limit must be an integer or None, not a boolean")
            if self.limit <= 0:
                raise ValueError("limit must be positive; use None for no limit")

        if not isinstance(self.sort, tuple):
            raise TypeError("sort must be a tuple of (field, direction) tuples")
        for pair in self.sort:
            if not isinstance(pair, tuple):
                raise TypeError("each sort entry must be a tuple")
            if len(pair) != 2:
                raise ValueError("each sort entry must contain a field and direction")
            name, direction = pair
            if not isinstance(name, str):
                raise TypeError("sort field names must be strings")
            if not name or "\0" in name:
                raise ValueError("sort field names must be nonempty without null characters")
            if type(direction) is not int:
                raise TypeError("sort directions must be integers, not booleans")
            if direction not in (1, -1):
                raise ValueError("sort directions must be 1 or -1")

    def to_json(self) -> str:
        """Serialize a version-1 JSON envelope with a Canonical Extended JSON filter.

        Metadata, projection, sort, skip and limit use ordinary JSON. Projections
        must contain JSON-serializable values. Preserves document and sort order.
        Dates normalize to millisecond precision
        and load as naive UTC datetimes. Binary subtype zero loads as bytes; large
        integers may load as bson.Int64. These preserve the serialized BSON value,
        not necessarily its original Python representation.

        Raises:
            TypeError: Structural validation or BSON-aware JSON encoding fails.
            ValueError: Validation fails or a document is ambiguous with an Extended
                JSON type wrapper. Use bson.Regex instead of a $regex/$options
                document. Unsupported Python values may raise codec exceptions.

        Examples:
            >>> query = MongoFindQuery(MongoCollectionRef("research", "businesses"), {})
            >>> MongoFindQuery.from_json(query.to_json()) == query
            True
        """
        self.validate()
        envelope = {
            "format_version": 1,
            "query_kind": "mongodb.find",
            "target": {"database": self.target.database, "collection": self.target.collection},
            "filter": self.filter,
            "projection": self.projection,
            "sort": self.sort,
            "skip": self.skip,
            "limit": self.limit,
        }
        return _dump_envelope(envelope, ("filter",))

    @classmethod
    def from_json(cls, text: str) -> Self:
        """Load a version-1 envelope produced by to_json.

        Whitespace and envelope field order may differ. BSON payloads must use
        Canonical Extended JSON. Unknown, missing and duplicate envelope fields
        are rejected; duplicate document keys are also rejected at any depth.

        Raises:
            TypeError: Input is not a string or a decoded field has an incorrect type.
            ValueError: JSON, the envelope, canonical payload or field values are invalid.
        """
        raw, data = _load_envelope(
            text, "mongodb.find", {"filter", "projection", "sort", "skip", "limit"}, ("filter",)
        )
        target = data["target"]
        sort = data["sort"]
        if not isinstance(sort, list) or any(not isinstance(pair, list) for pair in sort):
            raise TypeError("Serialized sort must be an array of arrays")

        query = cls(
            target=MongoCollectionRef(**target),
            filter=data["filter"],
            projection=data["projection"],
            sort=tuple(tuple(pair) for pair in sort),
            skip=data["skip"],
            limit=data["limit"],
        )
        _check_canonical_payload(raw, query.to_json(), ("filter",))
        return query

    def fingerprint(self) -> str:
        """Return the SHA-256 hex digest of the current serialized representation.

        Computed on every call after validation. Includes the format version,
        target and all query options; preserves document order. Identifies the
        serialized representation, not semantic equivalence or deployment state.
        Raises the same errors as to_json.
        """
        return hashlib.sha256(self.to_json().encode("utf-8")).hexdigest()

`fingerprint() -> str`

Return the SHA-256 hex digest of the current serialized representation.

Computed on every call after validation. Includes the format version, target and all query options; preserves document order. Identifies the serialized representation, not semantic equivalence or deployment state. Raises the same errors as to_json.

Source code in src/bondi/mongodb/query.py

def fingerprint(self) -> str:
    """Return the SHA-256 hex digest of the current serialized representation.

    Computed on every call after validation. Includes the format version,
    target and all query options; preserves document order. Identifies the
    serialized representation, not semantic equivalence or deployment state.
    Raises the same errors as to_json.
    """
    return hashlib.sha256(self.to_json().encode("utf-8")).hexdigest()

`from_json(text: str) -> Self` `classmethod`

Load a version-1 envelope produced by to_json.

Whitespace and envelope field order may differ. BSON payloads must use Canonical Extended JSON. Unknown, missing and duplicate envelope fields are rejected; duplicate document keys are also rejected at any depth.

Raises:

Type	Description
`TypeError`	Input is not a string or a decoded field has an incorrect type.
`ValueError`	JSON, the envelope, canonical payload or field values are invalid.

Source code in src/bondi/mongodb/query.py

@classmethod
def from_json(cls, text: str) -> Self:
    """Load a version-1 envelope produced by to_json.

    Whitespace and envelope field order may differ. BSON payloads must use
    Canonical Extended JSON. Unknown, missing and duplicate envelope fields
    are rejected; duplicate document keys are also rejected at any depth.

    Raises:
        TypeError: Input is not a string or a decoded field has an incorrect type.
        ValueError: JSON, the envelope, canonical payload or field values are invalid.
    """
    raw, data = _load_envelope(
        text, "mongodb.find", {"filter", "projection", "sort", "skip", "limit"}, ("filter",)
    )
    target = data["target"]
    sort = data["sort"]
    if not isinstance(sort, list) or any(not isinstance(pair, list) for pair in sort):
        raise TypeError("Serialized sort must be an array of arrays")

    query = cls(
        target=MongoCollectionRef(**target),
        filter=data["filter"],
        projection=data["projection"],
        sort=tuple(tuple(pair) for pair in sort),
        skip=data["skip"],
        limit=data["limit"],
    )
    _check_canonical_payload(raw, query.to_json(), ("filter",))
    return query

`to_json() -> str`

Serialize a version-1 JSON envelope with a Canonical Extended JSON filter.

Metadata, projection, sort, skip and limit use ordinary JSON. Projections must contain JSON-serializable values. Preserves document and sort order. Dates normalize to millisecond precision and load as naive UTC datetimes. Binary subtype zero loads as bytes; large integers may load as bson.Int64. These preserve the serialized BSON value, not necessarily its original Python representation.

Raises:

Type	Description
`TypeError`	Structural validation or BSON-aware JSON encoding fails.
`ValueError`	Validation fails or a document is ambiguous with an Extended JSON type wrapper. Use bson.Regex instead of a $regex/$options document. Unsupported Python values may raise codec exceptions.

Examples:

>>> query = MongoFindQuery(MongoCollectionRef("research", "businesses"), {})
>>> MongoFindQuery.from_json(query.to_json()) == query
True

Source code in src/bondi/mongodb/query.py

def to_json(self) -> str:
    """Serialize a version-1 JSON envelope with a Canonical Extended JSON filter.

    Metadata, projection, sort, skip and limit use ordinary JSON. Projections
    must contain JSON-serializable values. Preserves document and sort order.
    Dates normalize to millisecond precision
    and load as naive UTC datetimes. Binary subtype zero loads as bytes; large
    integers may load as bson.Int64. These preserve the serialized BSON value,
    not necessarily its original Python representation.

    Raises:
        TypeError: Structural validation or BSON-aware JSON encoding fails.
        ValueError: Validation fails or a document is ambiguous with an Extended
            JSON type wrapper. Use bson.Regex instead of a $regex/$options
            document. Unsupported Python values may raise codec exceptions.

    Examples:
        >>> query = MongoFindQuery(MongoCollectionRef("research", "businesses"), {})
        >>> MongoFindQuery.from_json(query.to_json()) == query
        True
    """
    self.validate()
    envelope = {
        "format_version": 1,
        "query_kind": "mongodb.find",
        "target": {"database": self.target.database, "collection": self.target.collection},
        "filter": self.filter,
        "projection": self.projection,
        "sort": self.sort,
        "skip": self.skip,
        "limit": self.limit,
    }
    return _dump_envelope(envelope, ("filter",))

`validate() -> None`

Check current structural constraints without modifying the query.

Raises:

Type	Description
`TypeError`	A field has an incorrect type.
`ValueError`	A field violates the constructor's value constraints.

BSON encoding and MongoDB operator validity are not checked here.

Source code in src/bondi/mongodb/query.py

def validate(self) -> None:
    """Check current structural constraints without modifying the query.

    Raises:
        TypeError: A field has an incorrect type.
        ValueError: A field violates the constructor's value constraints.

    BSON encoding and MongoDB operator validity are not checked here.
    """
    if not isinstance(self.target, MongoCollectionRef):
        raise TypeError("target must be a MongoCollectionRef")
    if not isinstance(self.filter, dict):
        raise TypeError("filter must be a dictionary")
    if self.projection is not None and not isinstance(self.projection, dict):
        raise TypeError("projection must be a dictionary or None")

    if type(self.skip) is not int:
        raise TypeError("skip must be an integer, not a boolean")
    if self.skip < 0:
        raise ValueError("skip must be nonnegative")
    if self.limit is not None:
        if type(self.limit) is not int:
            raise TypeError("limit must be an integer or None, not a boolean")
        if self.limit <= 0:
            raise ValueError("limit must be positive; use None for no limit")

    if not isinstance(self.sort, tuple):
        raise TypeError("sort must be a tuple of (field, direction) tuples")
    for pair in self.sort:
        if not isinstance(pair, tuple):
            raise TypeError("each sort entry must be a tuple")
        if len(pair) != 2:
            raise ValueError("each sort entry must contain a field and direction")
        name, direction = pair
        if not isinstance(name, str):
            raise TypeError("sort field names must be strings")
        if not name or "\0" in name:
            raise ValueError("sort field names must be nonempty without null characters")
        if type(direction) is not int:
            raise TypeError("sort directions must be integers, not booleans")
        if direction not in (1, -1):
            raise ValueError("sort directions must be 1 or -1")

`bondi.mongodb.MongoAggregateQuery` `dataclass`

Describe a collection-based aggregation without executing it.

Mutable and unhashable. The pipeline is deep-copied at construction. Later assignments and nested edits are neither copied nor automatically validated. Validation runs at construction, explicitly, and before serialization or fingerprinting. Only top-level stage structure is checked; stage semantics and nested pipelines are left to MongoDB. Write stages such as $out and $merge are representable: this class does not promise read-only execution.

Parameters:

Name	Type	Description	Default
`target`	`MongoCollectionRef`	Starting collection, independent of a deployment. References to additional collections remain in the native pipeline.	required
`pipeline`	`list[dict[str, Any]]`	Required ordered list of single-key stage dictionaries. An empty list is allowed. Each key must start with $ and contain a name after it. Values may be scalars, documents or arrays; their semantics are not checked.	required

Raises:

Type	Description
`TypeError`	Target, pipeline, a stage or a stage key has an incorrect type.
`ValueError`	A stage has other than one key or an invalid stage-name prefix.

Examples:

>>> stages = [{"$match": {"stars": {"$gte": 4}}}, {"$limit": 10}]
>>> query = MongoAggregateQuery(MongoCollectionRef("research", "businesses"), stages)
>>> query.pipeline[1]["$limit"] = 5
>>> stages[1]["$limit"]
10
>>> restored = MongoAggregateQuery.from_json(query.to_json())
>>> restored.fingerprint() == query.fingerprint()
True

Source code in src/bondi/mongodb/aggregate.py

@dataclass
class MongoAggregateQuery:
    """Describe a collection-based aggregation without executing it.

    Mutable and unhashable. The pipeline is deep-copied at construction. Later
    assignments and nested edits are neither copied nor automatically validated.
    Validation runs at construction, explicitly, and before serialization or
    fingerprinting. Only top-level stage structure is checked; stage semantics
    and nested pipelines are left to MongoDB. Write stages such as $out and $merge
    are representable: this class does not promise read-only execution.

    Args:
        target: Starting collection, independent of a deployment. References to
            additional collections remain in the native pipeline.
        pipeline: Required ordered list of single-key stage dictionaries. An empty
            list is allowed. Each key must start with $ and contain a name after it.
            Values may be scalars, documents or arrays; their semantics are not checked.

    Raises:
        TypeError: Target, pipeline, a stage or a stage key has an incorrect type.
        ValueError: A stage has other than one key or an invalid stage-name prefix.

    Examples:
        >>> stages = [{"$match": {"stars": {"$gte": 4}}}, {"$limit": 10}]
        >>> query = MongoAggregateQuery(MongoCollectionRef("research", "businesses"), stages)
        >>> query.pipeline[1]["$limit"] = 5
        >>> stages[1]["$limit"]
        10
        >>> restored = MongoAggregateQuery.from_json(query.to_json())
        >>> restored.fingerprint() == query.fingerprint()
        True
    """

    backend: ClassVar[str] = "mongodb"

    target: MongoCollectionRef
    pipeline: list[dict[str, Any]]

    def __post_init__(self) -> None:
        self.validate()
        self.pipeline = deepcopy(self.pipeline)

    def validate(self) -> None:
        """Check current target and top-level stage structure without mutation.

        Raises:
            TypeError: Target, pipeline, stage or stage key has an incorrect type.
            ValueError: A stage does not have exactly one $-prefixed nonempty name.

        Does not validate stage arguments, supported operators, nested pipelines,
        BSON encodability or whether a pipeline writes data.
        """
        if not isinstance(self.target, MongoCollectionRef):
            raise TypeError("target must be a MongoCollectionRef")
        if not isinstance(self.pipeline, list):
            raise TypeError("pipeline must be a list")
        for position, stage in enumerate(self.pipeline):
            if not isinstance(stage, dict):
                raise TypeError(f"pipeline stage {position} must be a dictionary")
            if len(stage) != 1:
                raise ValueError(f"pipeline stage {position} must have exactly one key")
            name = next(iter(stage))
            if not isinstance(name, str):
                raise TypeError(f"pipeline stage {position} key must be a string")
            if not name.startswith("$") or len(name) == 1:
                raise ValueError(
                    f"pipeline stage {position} key must start with $ followed by a name"
                )

    def to_json(self) -> str:
        """Serialize a validated version-1 mongodb.aggregate envelope.

        Uses ordinary JSON metadata and a Canonical Extended JSON pipeline,
        preserving stage and document order. Shares
        find-query BSON normalization: dates load as naive UTC with millisecond
        precision, binary subtype zero as bytes, and large integers may become
        bson.Int64. Literal documents ambiguous with Extended JSON wrappers are
        rejected. Serialization does not modify the query.

        Raises:
            TypeError: Structural validation or BSON-aware encoding fails.
            ValueError: Invalid structure or an ambiguous Extended JSON document.

        Unsupported values may also raise codec exceptions.
        """
        self.validate()
        return _dump_envelope(
            {
                "format_version": 1,
                "query_kind": "mongodb.aggregate",
                "target": {"database": self.target.database, "collection": self.target.collection},
                "pipeline": self.pipeline,
            },
            ("pipeline",),
        )

    @classmethod
    def from_json(cls, text: str) -> Self:
        """Load a version-1 aggregation envelope with a canonical BSON pipeline.

        Allows whitespace and envelope-field reordering. Rejects unknown versions,
        wrong query kinds, missing/extra fields and duplicate JSON keys at any depth.

        Raises:
            TypeError: Input or decoded query fields have incorrect types.
            ValueError: Invalid JSON, envelope, pipeline structure or noncanonical payload.
        """
        raw, data = _load_envelope(text, "mongodb.aggregate", {"pipeline"}, ("pipeline",))
        query = cls(MongoCollectionRef(**data["target"]), data["pipeline"])
        _check_canonical_payload(raw, query.to_json(), ("pipeline",))
        return query

    def fingerprint(self) -> str:
        """Compute SHA-256 of the current serialized representation after validation.

        Includes version, kind, target and pipeline; preserves stage and document
        order. Recomputed on every call. Does not identify semantic equivalence or
        deployment state. Raises the same errors as to_json.
        """
        return hashlib.sha256(self.to_json().encode("utf-8")).hexdigest()

`fingerprint() -> str`

Compute SHA-256 of the current serialized representation after validation.

Includes version, kind, target and pipeline; preserves stage and document order. Recomputed on every call. Does not identify semantic equivalence or deployment state. Raises the same errors as to_json.

Source code in src/bondi/mongodb/aggregate.py

def fingerprint(self) -> str:
    """Compute SHA-256 of the current serialized representation after validation.

    Includes version, kind, target and pipeline; preserves stage and document
    order. Recomputed on every call. Does not identify semantic equivalence or
    deployment state. Raises the same errors as to_json.
    """
    return hashlib.sha256(self.to_json().encode("utf-8")).hexdigest()

`from_json(text: str) -> Self` `classmethod`

Load a version-1 aggregation envelope with a canonical BSON pipeline.

Allows whitespace and envelope-field reordering. Rejects unknown versions, wrong query kinds, missing/extra fields and duplicate JSON keys at any depth.

Raises:

Type	Description
`TypeError`	Input or decoded query fields have incorrect types.
`ValueError`	Invalid JSON, envelope, pipeline structure or noncanonical payload.

Source code in src/bondi/mongodb/aggregate.py

@classmethod
def from_json(cls, text: str) -> Self:
    """Load a version-1 aggregation envelope with a canonical BSON pipeline.

    Allows whitespace and envelope-field reordering. Rejects unknown versions,
    wrong query kinds, missing/extra fields and duplicate JSON keys at any depth.

    Raises:
        TypeError: Input or decoded query fields have incorrect types.
        ValueError: Invalid JSON, envelope, pipeline structure or noncanonical payload.
    """
    raw, data = _load_envelope(text, "mongodb.aggregate", {"pipeline"}, ("pipeline",))
    query = cls(MongoCollectionRef(**data["target"]), data["pipeline"])
    _check_canonical_payload(raw, query.to_json(), ("pipeline",))
    return query

`to_json() -> str`

Serialize a validated version-1 mongodb.aggregate envelope.

Uses ordinary JSON metadata and a Canonical Extended JSON pipeline, preserving stage and document order. Shares find-query BSON normalization: dates load as naive UTC with millisecond precision, binary subtype zero as bytes, and large integers may become bson.Int64. Literal documents ambiguous with Extended JSON wrappers are rejected. Serialization does not modify the query.

Raises:

Type	Description
`TypeError`	Structural validation or BSON-aware encoding fails.
`ValueError`	Invalid structure or an ambiguous Extended JSON document.

Unsupported values may also raise codec exceptions.

Source code in src/bondi/mongodb/aggregate.py

def to_json(self) -> str:
    """Serialize a validated version-1 mongodb.aggregate envelope.

    Uses ordinary JSON metadata and a Canonical Extended JSON pipeline,
    preserving stage and document order. Shares
    find-query BSON normalization: dates load as naive UTC with millisecond
    precision, binary subtype zero as bytes, and large integers may become
    bson.Int64. Literal documents ambiguous with Extended JSON wrappers are
    rejected. Serialization does not modify the query.

    Raises:
        TypeError: Structural validation or BSON-aware encoding fails.
        ValueError: Invalid structure or an ambiguous Extended JSON document.

    Unsupported values may also raise codec exceptions.
    """
    self.validate()
    return _dump_envelope(
        {
            "format_version": 1,
            "query_kind": "mongodb.aggregate",
            "target": {"database": self.target.database, "collection": self.target.collection},
            "pipeline": self.pipeline,
        },
        ("pipeline",),
    )

`validate() -> None`

Check current target and top-level stage structure without mutation.

Raises:

Type	Description
`TypeError`	Target, pipeline, stage or stage key has an incorrect type.
`ValueError`	A stage does not have exactly one $-prefixed nonempty name.

Does not validate stage arguments, supported operators, nested pipelines, BSON encodability or whether a pipeline writes data.

Source code in src/bondi/mongodb/aggregate.py

def validate(self) -> None:
    """Check current target and top-level stage structure without mutation.

    Raises:
        TypeError: Target, pipeline, stage or stage key has an incorrect type.
        ValueError: A stage does not have exactly one $-prefixed nonempty name.

    Does not validate stage arguments, supported operators, nested pipelines,
    BSON encodability or whether a pipeline writes data.
    """
    if not isinstance(self.target, MongoCollectionRef):
        raise TypeError("target must be a MongoCollectionRef")
    if not isinstance(self.pipeline, list):
        raise TypeError("pipeline must be a list")
    for position, stage in enumerate(self.pipeline):
        if not isinstance(stage, dict):
            raise TypeError(f"pipeline stage {position} must be a dictionary")
        if len(stage) != 1:
            raise ValueError(f"pipeline stage {position} must have exactly one key")
        name = next(iter(stage))
        if not isinstance(name, str):
            raise TypeError(f"pipeline stage {position} key must be a string")
        if not name.startswith("$") or len(name) == 1:
            raise ValueError(
                f"pipeline stage {position} key must start with $ followed by a name"
            )

MongoDB queries

bondi.mongodb.MongoCollectionRef dataclass

bondi.mongodb.MongoFindQuery dataclass

fingerprint() -> str

from_json(text: str) -> Self classmethod

to_json() -> str

validate() -> None

bondi.mongodb.MongoAggregateQuery dataclass

fingerprint() -> str

from_json(text: str) -> Self classmethod

to_json() -> str

validate() -> None

`bondi.mongodb.MongoCollectionRef` `dataclass`

`bondi.mongodb.MongoFindQuery` `dataclass`

`fingerprint() -> str`

`from_json(text: str) -> Self` `classmethod`

`to_json() -> str`

`validate() -> None`

`bondi.mongodb.MongoAggregateQuery` `dataclass`

`fingerprint() -> str`

`from_json(text: str) -> Self` `classmethod`

`to_json() -> str`

`validate() -> None`