Skip to content

Couchbase queries

bondi.couchbase.CouchbaseQuery dataclass

Describe SQL++ text and bindings without parsing or executing the statement.

Mutable and unhashable. Parameter containers are deep-copied at construction. Later edits are not intercepted; validation runs explicitly, at construction, and before serialization or fingerprinting. Statement and context text are preserved exactly, including whitespace. No Couchbase SDK is required.

Parameters:

Name Type Description Default
statement str

Nonblank SQL++ text. Syntax, placeholder bindings and read-only behaviour are not checked, so invalid LLM candidates can be represented.

 required
named_parameters dict[str, Any]

Named JSON values, conventionally keyed without the $ placeholder prefix. Names are preserved, not parsed or normalized.

 dict()
positional_parameters list[Any]

Ordered JSON values for positional placeholders. Only one parameter style may be nonempty (a toolkit restriction).

 list()
query_context str | None

Optional nonblank native context string for name resolution. Its syntax is not parsed. None means no explicit context.

 None

Raises:

Type Description
TypeError

Fields have incorrect types or parameters contain non-JSON values.
ValueError

Text is blank, both parameter styles are used, or parameters contain nonfinite floats or circular references.

Examples:

>>> query = CouchbaseQuery("SELECT $minimum AS stars", named_parameters={"minimum": 4})
>>> restored = CouchbaseQuery.from_json(query.to_json())
>>> restored.fingerprint() == query.fingerprint()
True
Source code in src/bondi/couchbase/query.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@dataclass
class CouchbaseQuery:
    """Describe SQL++ text and bindings without parsing or executing the statement.

    Mutable and unhashable. Parameter containers are deep-copied at construction.
    Later edits are not intercepted; validation runs explicitly, at construction,
    and before serialization or fingerprinting. Statement and context text are
    preserved exactly, including whitespace. No Couchbase SDK is required.

    Args:
        statement: Nonblank SQL++ text. Syntax, placeholder bindings and read-only
            behaviour are not checked, so invalid LLM candidates can be represented.
        named_parameters: Named JSON values, conventionally keyed without the $
            placeholder prefix. Names are preserved, not parsed or normalized.
        positional_parameters: Ordered JSON values for positional placeholders.
            Only one parameter style may be nonempty (a toolkit restriction).
        query_context: Optional nonblank native context string for name resolution.
            Its syntax is not parsed. None means no explicit context.

    Raises:
        TypeError: Fields have incorrect types or parameters contain non-JSON values.
        ValueError: Text is blank, both parameter styles are used, or parameters
            contain nonfinite floats or circular references.

    Examples:
        >>> query = CouchbaseQuery("SELECT $minimum AS stars", named_parameters={"minimum": 4})
        >>> restored = CouchbaseQuery.from_json(query.to_json())
        >>> restored.fingerprint() == query.fingerprint()
        True
    """

    backend: ClassVar[str] = "couchbase"

    statement: str
    named_parameters: dict[str, Any] = field(default_factory=dict)
    positional_parameters: list[Any] = field(default_factory=list)
    query_context: str | None = None

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

    def validate(self) -> None:
        """Check text and JSON bindings without modifying them or parsing SQL++.

        Plain dict/list containers, string keys, strings, booleans, integers,
        finite floats and None are accepted. Tuples, BSON-specific values and
        dates are rejected rather than converted. JSON validity does not guarantee
        server acceptance, including numeric range or precision support.

        Raises:
            TypeError: A field or nested parameter has an incorrect type.
            ValueError: Blank text, mixed parameter styles, nonfinite numbers or cycles.
        """
        for name, value in (("statement", self.statement), ("query_context", self.query_context)):
            if name == "query_context" and value is None:
                continue
            if not isinstance(value, str):
                raise TypeError(f"{name} must be a string")
            if not value.strip():
                raise ValueError(f"{name} must not be blank")
        if type(self.named_parameters) is not dict:
            raise TypeError("named_parameters must be a dictionary")
        if type(self.positional_parameters) is not list:
            raise TypeError("positional_parameters must be a list")
        if self.named_parameters and self.positional_parameters:
            raise ValueError("Only one parameter style may be nonempty")
        _validate_json(self.named_parameters, set())
        _validate_json(self.positional_parameters, set())

    def to_json(self) -> str:
        """Serialize a validated version-1 couchbase.sqlpp envelope as standard JSON.

        Includes defaults; preserves exact statement/context text, array order
        and parameter dictionary insertion order. Does not use BSON Extended JSON.
        Raises the same validation errors as validate().
        """
        self.validate()
        return json.dumps(
            {
                "format_version": 1,
                "query_kind": "couchbase.sqlpp",
                "statement": self.statement,
                "named_parameters": self.named_parameters,
                "positional_parameters": self.positional_parameters,
                "query_context": self.query_context,
            },
            separators=(",", ":"),
            ensure_ascii=True,
            allow_nan=False,
        )

    @classmethod
    def from_json(cls, text: str) -> Self:
        """Load a version-1 SQL++ envelope and validate its fields.

        Rejects missing/extra fields, unknown versions or kinds, duplicate JSON
        keys, and nonfinite numeric values. Envelope order and formatting may vary;
        the exact statement string is retained.

        Raises:
            TypeError: Input or decoded fields have incorrect types.
            ValueError: Invalid JSON, envelope or query values.
        """
        if not isinstance(text, str):
            raise TypeError("text must be a string")

        def unique_object(pairs):
            result = {}
            for key, value in pairs:
                if key in result:
                    raise ValueError(f"Duplicate JSON key: {key}")
                result[key] = value
            return result

        def reject_constant(value):
            raise ValueError(f"Nonfinite JSON constant: {value}")

        data = json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_constant)
        fields = {
            "format_version",
            "query_kind",
            "statement",
            "named_parameters",
            "positional_parameters",
            "query_context",
        }
        if not isinstance(data, dict) or set(data) != fields:
            raise ValueError("Query envelope must contain exactly the version-1 fields")
        if type(data["format_version"]) is not int or data["format_version"] != 1:
            raise ValueError("Unsupported format_version")
        if data["query_kind"] != "couchbase.sqlpp":
            raise ValueError("Unsupported query_kind")
        return cls(
            statement=data["statement"],
            named_parameters=data["named_parameters"],
            positional_parameters=data["positional_parameters"],
            query_context=data["query_context"],
        )

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

        Includes statement text, bindings, context, kind and version. Whitespace
        and comments inside SQL++ affect the digest, as does parameter dictionary
        order. No semantic normalization or deployment identity is implied.
        Recomputed on each call; 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 validated serialized representation.

Includes statement text, bindings, context, kind and version. Whitespace and comments inside SQL++ affect the digest, as does parameter dictionary order. No semantic normalization or deployment identity is implied. Recomputed on each call; raises the same errors as to_json().

Source code in src/bondi/couchbase/query.py 
179
180
181
182
183
184
185
186
187
def fingerprint(self) -> str:
    """Compute SHA-256 of the current validated serialized representation.

    Includes statement text, bindings, context, kind and version. Whitespace
    and comments inside SQL++ affect the digest, as does parameter dictionary
    order. No semantic normalization or deployment identity is implied.
    Recomputed on each call; 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 SQL++ envelope and validate its fields.

Rejects missing/extra fields, unknown versions or kinds, duplicate JSON keys, and nonfinite numeric values. Envelope order and formatting may vary; the exact statement string is retained.

Raises:

Type Description
TypeError

Input or decoded fields have incorrect types.
ValueError

Invalid JSON, envelope or query values.
Source code in src/bondi/couchbase/query.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@classmethod
def from_json(cls, text: str) -> Self:
    """Load a version-1 SQL++ envelope and validate its fields.

    Rejects missing/extra fields, unknown versions or kinds, duplicate JSON
    keys, and nonfinite numeric values. Envelope order and formatting may vary;
    the exact statement string is retained.

    Raises:
        TypeError: Input or decoded fields have incorrect types.
        ValueError: Invalid JSON, envelope or query values.
    """
    if not isinstance(text, str):
        raise TypeError("text must be a string")

    def unique_object(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError(f"Duplicate JSON key: {key}")
            result[key] = value
        return result

    def reject_constant(value):
        raise ValueError(f"Nonfinite JSON constant: {value}")

    data = json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_constant)
    fields = {
        "format_version",
        "query_kind",
        "statement",
        "named_parameters",
        "positional_parameters",
        "query_context",
    }
    if not isinstance(data, dict) or set(data) != fields:
        raise ValueError("Query envelope must contain exactly the version-1 fields")
    if type(data["format_version"]) is not int or data["format_version"] != 1:
        raise ValueError("Unsupported format_version")
    if data["query_kind"] != "couchbase.sqlpp":
        raise ValueError("Unsupported query_kind")
    return cls(
        statement=data["statement"],
        named_parameters=data["named_parameters"],
        positional_parameters=data["positional_parameters"],
        query_context=data["query_context"],
    )

to_json() -> str

Serialize a validated version-1 couchbase.sqlpp envelope as standard JSON.

Includes defaults; preserves exact statement/context text, array order and parameter dictionary insertion order. Does not use BSON Extended JSON. Raises the same validation errors as validate().

Source code in src/bondi/couchbase/query.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def to_json(self) -> str:
    """Serialize a validated version-1 couchbase.sqlpp envelope as standard JSON.

    Includes defaults; preserves exact statement/context text, array order
    and parameter dictionary insertion order. Does not use BSON Extended JSON.
    Raises the same validation errors as validate().
    """
    self.validate()
    return json.dumps(
        {
            "format_version": 1,
            "query_kind": "couchbase.sqlpp",
            "statement": self.statement,
            "named_parameters": self.named_parameters,
            "positional_parameters": self.positional_parameters,
            "query_context": self.query_context,
        },
        separators=(",", ":"),
        ensure_ascii=True,
        allow_nan=False,
    )

validate() -> None

Check text and JSON bindings without modifying them or parsing SQL++.

Plain dict/list containers, string keys, strings, booleans, integers, finite floats and None are accepted. Tuples, BSON-specific values and dates are rejected rather than converted. JSON validity does not guarantee server acceptance, including numeric range or precision support.

Raises:

Type Description
TypeError

A field or nested parameter has an incorrect type.
ValueError

Blank text, mixed parameter styles, nonfinite numbers or cycles.
Source code in src/bondi/couchbase/query.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def validate(self) -> None:
    """Check text and JSON bindings without modifying them or parsing SQL++.

    Plain dict/list containers, string keys, strings, booleans, integers,
    finite floats and None are accepted. Tuples, BSON-specific values and
    dates are rejected rather than converted. JSON validity does not guarantee
    server acceptance, including numeric range or precision support.

    Raises:
        TypeError: A field or nested parameter has an incorrect type.
        ValueError: Blank text, mixed parameter styles, nonfinite numbers or cycles.
    """
    for name, value in (("statement", self.statement), ("query_context", self.query_context)):
        if name == "query_context" and value is None:
            continue
        if not isinstance(value, str):
            raise TypeError(f"{name} must be a string")
        if not value.strip():
            raise ValueError(f"{name} must not be blank")
    if type(self.named_parameters) is not dict:
        raise TypeError("named_parameters must be a dictionary")
    if type(self.positional_parameters) is not list:
        raise TypeError("positional_parameters must be a list")
    if self.named_parameters and self.positional_parameters:
        raise ValueError("Only one parameter style may be nonempty")
    _validate_json(self.named_parameters, set())
    _validate_json(self.positional_parameters, set())