Couchbase executors

bondi.couchbase.CouchbaseExecutor

Bases: _SyncExecutor

Synchronous database executor with an owned connection.

Create with from_uri(). Use with or explicitly call close() when finished. Calls retain one private event loop and must stay on the creating thread. Works in notebooks: the owned event loop runs on a dedicated background thread. Operation options, results, concurrency and errors match the async executor.

Source code in src/bondi/couchbase/sync_executor.py

class CouchbaseExecutor(_SyncExecutor):
    """Synchronous database executor with an owned connection.

    Create with from_uri(). Use with or explicitly call close() when finished.
    Calls retain one private event loop and must stay on the creating thread.
    Works in notebooks: the owned event loop runs on a dedicated background thread.
    Operation options, results, concurrency and errors match the async executor.
    """

    @classmethod
    def from_uri(
        cls,
        uri: str,
        *,
        username: str,
        password: str,
        concurrency: int = 1,
        continue_on_error: bool = False,
        timeout_seconds: float | None = None,
        cluster_kwargs: Mapping[str, Any] | None = None,
    ) -> Self:
        """Create an owned connection on a private event loop.

        Accepts the same connection options as the async executor.
        Close explicitly or use a with block.
        """

        async def connect():
            return await AsyncCouchbaseExecutor.from_uri(
                uri,
                username=username,
                password=password,
                concurrency=concurrency,
                continue_on_error=continue_on_error,
                timeout_seconds=timeout_seconds,
                cluster_kwargs=cluster_kwargs,
            )

        return cls._create(connect)

    def run(
        self,
        query: CouchbaseQuery,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> ExecutionResult:
        """Synchronously run; options match AsyncCouchbaseExecutor.run."""
        return self._call("run", query, collect_results=collect_results, query_kwargs=query_kwargs)

    def explain(
        self, query: CouchbaseQuery, *, query_kwargs: Mapping[str, Any] | None = None
    ) -> ExecutionResult:
        """Synchronously explain; options match AsyncCouchbaseExecutor.explain."""
        return self._call("explain", query, query_kwargs=query_kwargs)

    def profile(
        self,
        query: CouchbaseQuery,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> ExecutionResult:
        """Synchronously profile; options match AsyncCouchbaseExecutor.profile."""
        return self._call(
            "profile", query, collect_results=collect_results, query_kwargs=query_kwargs
        )

    def run_workload(
        self,
        workload: Workload,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> list[ExecutionResult | None]:
        """Synchronously run workload; options match AsyncCouchbaseExecutor.run_workload."""
        return self._call(
            "run_workload", workload, collect_results=collect_results, query_kwargs=query_kwargs
        )

    def explain_workload(
        self, workload: Workload, *, query_kwargs: Mapping[str, Any] | None = None
    ) -> list[ExecutionResult | None]:
        """Synchronously explain workload; options match AsyncCouchbaseExecutor.explain_workload."""
        return self._call("explain_workload", workload, query_kwargs=query_kwargs)

    def profile_workload(
        self,
        workload: Workload,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> list[ExecutionResult | None]:
        """Synchronously profile workload; options match AsyncCouchbaseExecutor.profile_workload."""
        return self._call(
            "profile_workload", workload, collect_results=collect_results, query_kwargs=query_kwargs
        )

close()

Close the connection, stop its loop and join its thread; idempotent.

Source code in src/bondi/_sync.py

def close(self) -> None:
    """Close the connection, stop its loop and join its thread; idempotent."""
    if self._closed:
        return
    self._check()
    try:
        self._submit(self._executor.close, cancel_on_interrupt=False)
    finally:
        self._shutdown()

explain(query, *, query_kwargs=None)

Synchronously explain; options match AsyncCouchbaseExecutor.explain.

Source code in src/bondi/couchbase/sync_executor.py

def explain(
    self, query: CouchbaseQuery, *, query_kwargs: Mapping[str, Any] | None = None
) -> ExecutionResult:
    """Synchronously explain; options match AsyncCouchbaseExecutor.explain."""
    return self._call("explain", query, query_kwargs=query_kwargs)

explain_workload(workload, *, query_kwargs=None)

Synchronously explain workload; options match AsyncCouchbaseExecutor.explain_workload.

Source code in src/bondi/couchbase/sync_executor.py

def explain_workload(
    self, workload: Workload, *, query_kwargs: Mapping[str, Any] | None = None
) -> list[ExecutionResult | None]:
    """Synchronously explain workload; options match AsyncCouchbaseExecutor.explain_workload."""
    return self._call("explain_workload", workload, query_kwargs=query_kwargs)

from_uri(uri, *, username, password, concurrency=1, continue_on_error=False, timeout_seconds=None, cluster_kwargs=None) classmethod

Create an owned connection on a private event loop.

Accepts the same connection options as the async executor. Close explicitly or use a with block.

Source code in src/bondi/couchbase/sync_executor.py

@classmethod
def from_uri(
    cls,
    uri: str,
    *,
    username: str,
    password: str,
    concurrency: int = 1,
    continue_on_error: bool = False,
    timeout_seconds: float | None = None,
    cluster_kwargs: Mapping[str, Any] | None = None,
) -> Self:
    """Create an owned connection on a private event loop.

    Accepts the same connection options as the async executor.
    Close explicitly or use a with block.
    """

    async def connect():
        return await AsyncCouchbaseExecutor.from_uri(
            uri,
            username=username,
            password=password,
            concurrency=concurrency,
            continue_on_error=continue_on_error,
            timeout_seconds=timeout_seconds,
            cluster_kwargs=cluster_kwargs,
        )

    return cls._create(connect)

profile(query, *, collect_results=False, query_kwargs=None)

Synchronously profile; options match AsyncCouchbaseExecutor.profile.

Source code in src/bondi/couchbase/sync_executor.py

def profile(
    self,
    query: CouchbaseQuery,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> ExecutionResult:
    """Synchronously profile; options match AsyncCouchbaseExecutor.profile."""
    return self._call(
        "profile", query, collect_results=collect_results, query_kwargs=query_kwargs
    )

profile_workload(workload, *, collect_results=False, query_kwargs=None)

Synchronously profile workload; options match AsyncCouchbaseExecutor.profile_workload.

Source code in src/bondi/couchbase/sync_executor.py

def profile_workload(
    self,
    workload: Workload,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> list[ExecutionResult | None]:
    """Synchronously profile workload; options match AsyncCouchbaseExecutor.profile_workload."""
    return self._call(
        "profile_workload", workload, collect_results=collect_results, query_kwargs=query_kwargs
    )

run(query, *, collect_results=False, query_kwargs=None)

Synchronously run; options match AsyncCouchbaseExecutor.run.

Source code in src/bondi/couchbase/sync_executor.py

def run(
    self,
    query: CouchbaseQuery,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> ExecutionResult:
    """Synchronously run; options match AsyncCouchbaseExecutor.run."""
    return self._call("run", query, collect_results=collect_results, query_kwargs=query_kwargs)

run_workload(workload, *, collect_results=False, query_kwargs=None)

Synchronously run workload; options match AsyncCouchbaseExecutor.run_workload.

Source code in src/bondi/couchbase/sync_executor.py

def run_workload(
    self,
    workload: Workload,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> list[ExecutionResult | None]:
    """Synchronously run workload; options match AsyncCouchbaseExecutor.run_workload."""
    return self._call(
        "run_workload", workload, collect_results=collect_results, query_kwargs=query_kwargs
    )

bondi.couchbase.AsyncCouchbaseExecutor

Run, explain and profile SQL++ queries with bounded async workload workers.

Parameters:

Name	Type	Description	Default
cluster	Cluster	Borrowed acouchbase Cluster; the caller closes it.	required
concurrency	int	Positive maximum in-flight entries per workload call.	1
continue_on_error	bool	Continue scheduling after database/timeout failures.	False
timeout_seconds	float | None	Positive finite per-operation deadline, including row consumption. None retains the SDK's configured query timeout.	None

Install the couchbase extra to instantiate this class. Inputs are retained by reference and must not be modified during calls. Use one event loop per cluster. Single operations raise; workloads capture SDK operation errors. Programming errors and cancellation propagate. No additional retries or write guard are added.

Source code in src/bondi/couchbase/executor.py

class AsyncCouchbaseExecutor:
    """Run, explain and profile SQL++ queries with bounded async workload workers.

    Args:
        cluster: Borrowed acouchbase Cluster; the caller closes it.
        concurrency: Positive maximum in-flight entries per workload call.
        continue_on_error: Continue scheduling after database/timeout failures.
        timeout_seconds: Positive finite per-operation deadline, including row
            consumption. None retains the SDK's configured query timeout.

    Install the couchbase extra to instantiate this class. Inputs are retained by
    reference and must not be modified during calls. Use one event loop per cluster.
    Single operations raise; workloads capture SDK operation errors. Programming
    errors and cancellation propagate. No additional retries or write guard are added.
    """

    def __init__(
        self,
        cluster: Cluster,
        *,
        concurrency: int = 1,
        continue_on_error: bool = False,
        timeout_seconds: float | None = None,
    ) -> None:
        try:
            import acouchbase.cluster  # noqa: F401
        except ModuleNotFoundError as error:
            if error.name not in ("acouchbase", "couchbase", "acouchbase.cluster"):
                raise
            raise ImportError(
                "Install Couchbase support with: uv sync --extra couchbase"
            ) from error
        if type(concurrency) is not int or concurrency < 1:
            raise ValueError("concurrency must be a positive integer")
        if type(continue_on_error) is not bool:
            raise TypeError("continue_on_error must be a boolean")
        if timeout_seconds is not None and (
            type(timeout_seconds) not in (int, float)
            or not math.isfinite(timeout_seconds)
            or timeout_seconds <= 0
        ):
            raise ValueError("timeout_seconds must be positive and finite, or None")
        self.cluster = cluster
        self.concurrency = concurrency
        self.continue_on_error = continue_on_error
        self.timeout_seconds = timeout_seconds
        self._owns_cluster = False
        self._closed = False

    @classmethod
    async def from_uri(
        cls,
        uri: str,
        *,
        username: str,
        password: str,
        concurrency: int = 1,
        continue_on_error: bool = False,
        timeout_seconds: float | None = None,
        cluster_kwargs: Mapping[str, Any] | None = None,
    ) -> Self:
        """Connect an owned cluster and wait up to 30 seconds for the Query service.

        Await this factory before using async with. cluster_kwargs supplies native
        ClusterOptions; authentication comes from username/password. On readiness
        failure or cancellation the newly connected cluster is closed. Connection
        setup is outside per-query timing and uses SDK connection timeout settings.
        """
        executor = cls(
            None,
            concurrency=concurrency,
            continue_on_error=continue_on_error,
            timeout_seconds=timeout_seconds,
        )
        from acouchbase.cluster import Cluster
        from couchbase.auth import PasswordAuthenticator
        from couchbase.diagnostics import ServiceType
        from couchbase.options import ClusterOptions, WaitUntilReadyOptions

        options = _options(cluster_kwargs, {"authenticator"})
        executor.cluster = await Cluster.connect(
            uri, ClusterOptions(PasswordAuthenticator(username, password), **options)
        )
        executor._owns_cluster = True
        try:
            await executor.cluster.wait_until_ready(
                timedelta(seconds=30), WaitUntilReadyOptions(service_types=[ServiceType.Query])
            )
        except BaseException:
            await executor.close()
            raise
        return executor

    async def close(self) -> None:
        """Close an owned cluster once; leave borrowed clusters open."""
        if self._owns_cluster and not self._closed:
            await self.cluster.close()
            self._closed = True

    async def __aenter__(self) -> Self:
        self._check_open()
        return self

    async def __aexit__(self, exc_type, exc, traceback) -> None:
        await self.close()

    def _check_open(self) -> None:
        if self._closed:
            raise RuntimeError("Executor is closed")

    def _validate_query(self, query: CouchbaseQuery) -> None:
        self._check_open()
        if not isinstance(query, CouchbaseQuery):
            raise TypeError("Expected CouchbaseQuery")
        query.validate()

    @staticmethod
    def _collect(value: bool) -> None:
        if type(value) is not bool:
            raise TypeError("collect_results must be a boolean")

    async def run(
        self,
        query: CouchbaseQuery,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> ExecutionResult:
        """Consume all rows, optionally retain them, and return metrics in raw.

        query_kwargs accepts native QueryOptions such as scan_consistency or
        read_only. Bindings, context, timeout, metrics, profile, raw and serializer
        are reserved. SDK consistency defaults apply unless explicitly overridden.
        """
        return await self._execute(query, "run", collect_results, query_kwargs)

    async def explain(
        self, query: CouchbaseQuery, *, query_kwargs: Mapping[str, Any] | None = None
    ) -> ExecutionResult:
        """Prefix the supplied statement with EXPLAIN and return its native plan in raw.

        Supply the original SQL++ statement, without an EXPLAIN prefix. No execution
        statistics are collected for that statement. row_count/documents remain None;
        EXPLAIN request metrics are not measurements of the underlying query.
        """
        return await self._execute(query, "explain", False, query_kwargs)

    async def profile(
        self,
        query: CouchbaseQuery,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> ExecutionResult:
        """Execute with TIMINGS profiling, consuming all rows and returning metadata.

        Unlike MongoDB profiling, Couchbase can also return the actual query rows.
        When executionTimings is absent, retain basic metrics and add a warning.
        No extra EXPLAIN request or automatic summary parsing is performed.
        """
        return await self._execute(query, "profile", collect_results, query_kwargs)

    async def _execute(
        self, query, mode: Literal["run", "explain", "profile"], collect_results, query_kwargs
    ) -> ExecutionResult:
        from couchbase.n1ql import QueryProfile
        from couchbase.options import QueryOptions

        self._validate_query(query)
        self._collect(collect_results)
        options = _options(query_kwargs, _RESERVED)
        options.update(
            named_parameters=query.named_parameters,
            positional_parameters=query.positional_parameters,
            query_context=query.query_context,
            metrics=True,
            profile=QueryProfile.TIMINGS if mode == "profile" else QueryProfile.OFF,
        )
        if self.timeout_seconds is not None:
            options["timeout"] = timedelta(seconds=self.timeout_seconds)
        statement = "EXPLAIN " + query.statement if mode == "explain" else query.statement
        rows = [] if collect_results or mode == "explain" else None
        count = 0
        start = perf_counter()
        # SDK cancellation finalizes its async streaming iterator. A local deadline
        # also bounds time spent consuming rows; the SDK timeout bounds server work.
        async with asyncio.timeout(self.timeout_seconds):
            response = self.cluster.query(statement, QueryOptions(**options))
            async for row in response.rows():
                count += 1
                if rows is not None:
                    rows.append(row)
            raw = _metadata(response.metadata())
        warnings = tuple(f"{w['code']}: {w['message']}" for w in raw["warnings"])
        if mode == "profile" and not (raw["profile"] or {}).get("executionTimings"):
            warnings += (
                "Detailed profiling unavailable: server did not return executionTimings. "
                "Basic metrics are retained; detailed profiling requires Enterprise support.",
            )
        if mode == "explain":
            if len(rows) != 1 or not isinstance(rows[0], dict):
                raise ValueError("Expected one plan document from Couchbase EXPLAIN")
            raw = rows[0]
        return ExecutionResult(
            (perf_counter() - start) * 1000,
            row_count=None if mode == "explain" else count,
            documents=rows if collect_results else None,
            raw=raw,
            warnings=warnings,
        )

    async def run_workload(
        self,
        workload: Workload,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> list[ExecutionResult | None]:
        """Run entries with input-aligned results and bounded concurrency.

        None marks unstarted entries. With continue_on_error=False, stop scheduling
        after observing a failure and drain in-flight operations. True attempts all
        entries. Partial rows are discarded on failure. Validate all inputs first.
        """
        self._collect(collect_results)
        options = _options(query_kwargs, _RESERVED)
        return await self._workload(
            workload,
            lambda query: self.run(query, collect_results=collect_results, query_kwargs=options),
        )

    async def explain_workload(
        self, workload: Workload, *, query_kwargs: Mapping[str, Any] | None = None
    ) -> list[ExecutionResult | None]:
        """Explain entries with run_workload's scheduling and failure policy."""
        options = _options(query_kwargs, _RESERVED)
        return await self._workload(
            workload, lambda query: self.explain(query, query_kwargs=options)
        )

    async def profile_workload(
        self,
        workload: Workload,
        *,
        collect_results: bool = False,
        query_kwargs: Mapping[str, Any] | None = None,
    ) -> list[ExecutionResult | None]:
        """Profile entries with run_workload's scheduling and failure policy."""
        self._collect(collect_results)
        options = _options(query_kwargs, _RESERVED)
        return await self._workload(
            workload,
            lambda query: self.profile(
                query, collect_results=collect_results, query_kwargs=options
            ),
        )

    async def _workload(self, workload, operation) -> list[ExecutionResult | None]:
        from couchbase.exceptions import (
            AlreadyQueriedException,
            CouchbaseException,
            InvalidArgumentException,
            InvalidValueException,
        )

        self._check_open()
        if not isinstance(workload, Workload):
            raise TypeError("workload must be a Workload")
        workload.validate()
        for entry in workload:
            self._validate_query(entry.query)
        results: list[ExecutionResult | None] = [None] * len(workload)
        next_index = 0
        stopped = False

        async def worker():
            nonlocal next_index, stopped
            while not stopped and next_index < len(workload):
                index = next_index
                next_index += 1
                start = perf_counter()
                try:
                    results[index] = await operation(workload[index].query)
                except (InvalidArgumentException, InvalidValueException, AlreadyQueriedException):
                    raise  # Invalid SDK options are API misuse, not database outcomes.
                except (CouchbaseException, TimeoutError) as error:
                    results[index] = ExecutionResult((perf_counter() - start) * 1000, error=error)
                    if not self.continue_on_error:
                        stopped = True

        workers = [
            asyncio.create_task(worker()) for _ in range(min(self.concurrency, len(workload)))
        ]
        try:
            await asyncio.gather(*workers)
        except BaseException:
            for task in workers:
                task.cancel()
            await asyncio.gather(*workers, return_exceptions=True)
            raise
        return results

close() async

Close an owned cluster once; leave borrowed clusters open.

Source code in src/bondi/couchbase/executor.py

async def close(self) -> None:
    """Close an owned cluster once; leave borrowed clusters open."""
    if self._owns_cluster and not self._closed:
        await self.cluster.close()
        self._closed = True

explain(query, *, query_kwargs=None) async

Prefix the supplied statement with EXPLAIN and return its native plan in raw.

Supply the original SQL++ statement, without an EXPLAIN prefix. No execution statistics are collected for that statement. row_count/documents remain None; EXPLAIN request metrics are not measurements of the underlying query.

Source code in src/bondi/couchbase/executor.py

async def explain(
    self, query: CouchbaseQuery, *, query_kwargs: Mapping[str, Any] | None = None
) -> ExecutionResult:
    """Prefix the supplied statement with EXPLAIN and return its native plan in raw.

    Supply the original SQL++ statement, without an EXPLAIN prefix. No execution
    statistics are collected for that statement. row_count/documents remain None;
    EXPLAIN request metrics are not measurements of the underlying query.
    """
    return await self._execute(query, "explain", False, query_kwargs)

explain_workload(workload, *, query_kwargs=None) async

Explain entries with run_workload's scheduling and failure policy.

Source code in src/bondi/couchbase/executor.py

async def explain_workload(
    self, workload: Workload, *, query_kwargs: Mapping[str, Any] | None = None
) -> list[ExecutionResult | None]:
    """Explain entries with run_workload's scheduling and failure policy."""
    options = _options(query_kwargs, _RESERVED)
    return await self._workload(
        workload, lambda query: self.explain(query, query_kwargs=options)
    )

from_uri(uri, *, username, password, concurrency=1, continue_on_error=False, timeout_seconds=None, cluster_kwargs=None) async classmethod

Connect an owned cluster and wait up to 30 seconds for the Query service.

Await this factory before using async with. cluster_kwargs supplies native ClusterOptions; authentication comes from username/password. On readiness failure or cancellation the newly connected cluster is closed. Connection setup is outside per-query timing and uses SDK connection timeout settings.

Source code in src/bondi/couchbase/executor.py

@classmethod
async def from_uri(
    cls,
    uri: str,
    *,
    username: str,
    password: str,
    concurrency: int = 1,
    continue_on_error: bool = False,
    timeout_seconds: float | None = None,
    cluster_kwargs: Mapping[str, Any] | None = None,
) -> Self:
    """Connect an owned cluster and wait up to 30 seconds for the Query service.

    Await this factory before using async with. cluster_kwargs supplies native
    ClusterOptions; authentication comes from username/password. On readiness
    failure or cancellation the newly connected cluster is closed. Connection
    setup is outside per-query timing and uses SDK connection timeout settings.
    """
    executor = cls(
        None,
        concurrency=concurrency,
        continue_on_error=continue_on_error,
        timeout_seconds=timeout_seconds,
    )
    from acouchbase.cluster import Cluster
    from couchbase.auth import PasswordAuthenticator
    from couchbase.diagnostics import ServiceType
    from couchbase.options import ClusterOptions, WaitUntilReadyOptions

    options = _options(cluster_kwargs, {"authenticator"})
    executor.cluster = await Cluster.connect(
        uri, ClusterOptions(PasswordAuthenticator(username, password), **options)
    )
    executor._owns_cluster = True
    try:
        await executor.cluster.wait_until_ready(
            timedelta(seconds=30), WaitUntilReadyOptions(service_types=[ServiceType.Query])
        )
    except BaseException:
        await executor.close()
        raise
    return executor

profile(query, *, collect_results=False, query_kwargs=None) async

Execute with TIMINGS profiling, consuming all rows and returning metadata.

Unlike MongoDB profiling, Couchbase can also return the actual query rows. When executionTimings is absent, retain basic metrics and add a warning. No extra EXPLAIN request or automatic summary parsing is performed.

Source code in src/bondi/couchbase/executor.py

async def profile(
    self,
    query: CouchbaseQuery,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> ExecutionResult:
    """Execute with TIMINGS profiling, consuming all rows and returning metadata.

    Unlike MongoDB profiling, Couchbase can also return the actual query rows.
    When executionTimings is absent, retain basic metrics and add a warning.
    No extra EXPLAIN request or automatic summary parsing is performed.
    """
    return await self._execute(query, "profile", collect_results, query_kwargs)

profile_workload(workload, *, collect_results=False, query_kwargs=None) async

Profile entries with run_workload's scheduling and failure policy.

Source code in src/bondi/couchbase/executor.py

async def profile_workload(
    self,
    workload: Workload,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> list[ExecutionResult | None]:
    """Profile entries with run_workload's scheduling and failure policy."""
    self._collect(collect_results)
    options = _options(query_kwargs, _RESERVED)
    return await self._workload(
        workload,
        lambda query: self.profile(
            query, collect_results=collect_results, query_kwargs=options
        ),
    )

run(query, *, collect_results=False, query_kwargs=None) async

Consume all rows, optionally retain them, and return metrics in raw.

query_kwargs accepts native QueryOptions such as scan_consistency or read_only. Bindings, context, timeout, metrics, profile, raw and serializer are reserved. SDK consistency defaults apply unless explicitly overridden.

Source code in src/bondi/couchbase/executor.py

async def run(
    self,
    query: CouchbaseQuery,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> ExecutionResult:
    """Consume all rows, optionally retain them, and return metrics in raw.

    query_kwargs accepts native QueryOptions such as scan_consistency or
    read_only. Bindings, context, timeout, metrics, profile, raw and serializer
    are reserved. SDK consistency defaults apply unless explicitly overridden.
    """
    return await self._execute(query, "run", collect_results, query_kwargs)

run_workload(workload, *, collect_results=False, query_kwargs=None) async

Run entries with input-aligned results and bounded concurrency.

None marks unstarted entries. With continue_on_error=False, stop scheduling after observing a failure and drain in-flight operations. True attempts all entries. Partial rows are discarded on failure. Validate all inputs first.

Source code in src/bondi/couchbase/executor.py

async def run_workload(
    self,
    workload: Workload,
    *,
    collect_results: bool = False,
    query_kwargs: Mapping[str, Any] | None = None,
) -> list[ExecutionResult | None]:
    """Run entries with input-aligned results and bounded concurrency.

    None marks unstarted entries. With continue_on_error=False, stop scheduling
    after observing a failure and drain in-flight operations. True attempts all
    entries. Partial rows are discarded on failure. Validate all inputs first.
    """
    self._collect(collect_results)
    options = _options(query_kwargs, _RESERVED)
    return await self._workload(
        workload,
        lambda query: self.run(query, collect_results=collect_results, query_kwargs=options),
    )

Workloads, failures and options

from bondi.workload import Workload, WorkloadEntry

workload = Workload([WorkloadEntry(query), WorkloadEntry(query)])
# Inside the executor context:
results = executor.run_workload(workload)
plans = executor.explain_workload(workload)
profiles = executor.profile_workload(workload)

All methods share bounded concurrency, input ordering and failure policy. With continue_on_error=False, scheduling stops after observing an error while in-flight operations finish. Unstarted entries are None. With True, remaining entries are attempted. Expected SDK failures are stored in .error; single-query methods raise. Partial results are discarded. SDK argument errors, programming errors and caller cancellation propagate after workers are cleaned up. Timeout status recognizes both ambiguous and unambiguous SDK timeouts and local deadlines.

timeout_seconds sets both an SDK query timeout and a local deadline through row consumption. None leaves the SDK timeout in effect. Inputs are not snapshotted; do not mutate them during operations.

query_kwargs passes native QueryOptions, including scan_consistency, read_only, adhoc and client_context_id. SDK defaults apply unless overridden; in particular, NOT_BOUNDED may observe indexes before recent writes arrive. Use REQUEST_PLUS when fresh index contents matter, and account for that wait in measurements. Bindings, query context, metrics, profile, timeout and serializer are reserved. The nested SDK raw escape hatch is also reserved because it can bypass these controls. No read-only guard or extra retry policy is imposed by the toolkit.

See the official SDK query guide for native options. The implementation is tested with SDK 4.6.3 and both local Couchbase 7.6.2 editions.