Sync Query Builders¶

class aerospike_sdk.sync.operations.query.QueryBuilder[source]¶

Bases: _QueryBuilderBase, _BlockingQueryDispatch, _WriteVerbs[WriteSegmentBuilder]

Synchronous query builder.

Inherits state + chaining from _QueryBuilderBase and the blocking-IO dispatchers from _BlockingQueryDispatch. Provides sync execute() that routes through Tier 1 (fast path / multi-key list dispatch), Tier 1b (multi-spec blocking dispatch), or Tier 2 (dataset / SI / scan streaming) using PNC _blocking entries. No asyncio loop involved.

Multi-key chains are split into per-node sub-batches, and a node whose sub-batch holds a single key is sent a regular single-record command instead of a batch request — automatically and per-node, so size-1 batches need no special-casing by the caller.

bin(bin_name)[source]¶

Open a per-bin read builder targeting this query builder.

Return type:: QueryBinBuilder[QueryBuilder]

execute(on_error=None)[source]¶

Run the configured query/write chain synchronously.

Tier 1: single-key + multi-key + all op-types (returns list). Tier 1b: multi-spec sequential dispatch via PNC batch_blocking. Tier 2: dataset / SI / scan streams (returns Recordset; lazy).

Return type:: RecordStream

first(on_error=None)[source]¶

Execute and return the first row, or None when there are none.

Reading one record by key is the most common thing this client does, and going through execute() costs a second call for it. This is that pair in one; the stream is closed either way.

Semantics come from RecordStream.first(), not from a second set of rules: per-record failures arrive as data (is_ok=False), while cluster-level errors raise.

Parameters:: on_error (Union[ErrorStrategy, Callable[[Key, int, AerospikeError], None], None]) – Per-operation error handling, as for execute().
Return type:: Optional[RecordResult]
Returns:: The first RecordResult, or None when the query matched nothing.

Example:

result = session.query(key).first()
if result is not None and result.is_ok:
    print(result.record.bins)

See also

first_or_raise(): Raise instead of returning None.

first_or_raise(on_error=None)[source]¶

Execute and return the first row, requiring it to exist and be OK.

Semantics come from RecordStream.first_or_raise().

Parameters:

on_error (Union[ErrorStrategy, Callable[[Key, int, AerospikeError], None], None]) – Per-operation error handling, as for execute().

Return type:

RecordResult

Returns:

The first successful RecordResult.

Raises:

RecordNotFoundError – The query matched no records.
AerospikeError – The first row reported a failure.

Example:

record = session.query(key).first_or_raise().record_or_raise()

See also

first(): Returns None on an empty result instead.

stream(on_error=None)[source]¶

Execute lazily — results stream back as each node responds.

The streaming counterpart to execute(). Where execute() materializes every result before returning (writes complete on return), this dispatches the key-batch through PNC’s blocking batch_stream and yields each row as its node responds — the first results are available as soon as the first node responds, without waiting for the rest, and peak memory stays bounded to the in-flight node responses. Results arrive in completion order, not input order; use RecordResult.index to correlate.

No writes-complete-on-return guarantee. A caller that never drains the returned stream may leave writes in-flight. For writes-done-on-return semantics use execute() (buffered).

Parameters:: on_error (Union[ErrorStrategy, Callable[[Key, int, AerospikeError], None], None]) – A (key, index, exception) -> None callback for per-key failures (dispatched as records arrive and excluded from the stream). An ErrorStrategy enum collapses to inline errors (the stream default).
Return type:: RecordStream
Returns:: A lazy RecordStream.

__init__(client, namespace, set_name, behavior, cached_read_policy=None, cached_write_policy=None, cached_read_policy_sc=None, cached_write_policy_sc=None, cached_read_operate_policy=None, cached_read_operate_policy_sc=None, txn=None, namespace_mode_resolver=None, namespace_mode_resolver_blocking=None, sdk_client=None, supports_server_compiled_ael=None, supports_query_selection=None)¶

Initialize a QueryBuilder.

Parameters:

client (Client) – The underlying async client.
namespace (str) – The namespace name.
set_name (str) – The set name.
behavior (Behavior) – The Behavior every policy this builder sends is derived from.
cached_read_policy (Optional[ReadPolicy]) – Pre-computed read policy from the session.
cached_write_policy (Optional[WritePolicy]) – Pre-computed write policy from the session.
cached_read_operate_policy (Optional[WritePolicy]) – Pre-computed policy for single-key operate calls that only read, from the session.
txn (Optional[Txn]) – Optional active Txn captured from a transactional session at construction; every policy this builder hands to the PNC gets stamped with it. None means no transaction participation. Callers rarely pass this directly — transactional sessions thread it through automatically.
namespace_mode_resolver (Optional[Callable[[str], Awaitable[Mode]]]) – Optional async callable namespace -> Mode used to apply AP vs SC behavior scopes before policies are built. Session-scoped builders supply this; client-only builders omit it.
namespace_mode_resolver_blocking (Optional[Callable[[str], Mode]]) – Sync counterpart used by execute_blocking() (the sync path bypassing asyncio).
sdk_client (Optional[Any]) – Optional owning SDK client, consulted at execute time for SDK-level settings (implicit batch-write transactions) and cluster MRT capability. None disables implicit transactions for this builder.

add_operation(op)¶

Append a read operation. Returns self so calls can chain.

Return type:: Self

bins(*bin_names)¶

Restrict the read to a non-empty set of bin names.

Accepts the names as varargs or as a single sequence, matching with_op_projection() and the other varargs builders.

Mutually exclusive with with_no_bins().

Parameters:

*bin_names (Union[str, Sequence[str]]) – Bin names to return, either as separate arguments or as one sequence of names.

Return type:

Self

Returns:

This builder for method chaining.

Raises:

TypeError – If a sequence is mixed with further positional names.
ValueError – If no bin names are given or with_no_bins() was already called.

Example:

Restrict a query or key read to specific bins::

    stream = await session.query(users.id(1)).bins("name", "email").execute()

See also

with_no_bins(): Metadata-only reads without bin payloads. bin(): Per-bin operations (CDT, expressions).

chunk_size(chunk_size)¶

Tune server-side streaming chunk size (maps to query policy max_records chunking).

This method controls how many records are fetched per chunk from the server when using server-side streaming. The chunk size affects memory usage and network round trips (Larger values reduce round trips; smaller values bound memory per fetch). This is distinct from client-side pagination.

Important

Advancing between chunks is the caller’s job. Iterating the stream yields only the chunk already loaded, so a plain for / async for over a chunked query stops after chunk_size records and the rest are never fetched. Drive has_more_chunks() in an outer loop to read the whole result, as the example below does.

Parameters:: chunk_size (int) – Records per chunk; must be positive.
Return type:: Self
Returns:: This builder for chaining.
Raises:: ValueError – If chunk_size <= 0.

Example:

stream = await session.query(dataset).chunk_size(100).execute()
try:
    while await stream.has_more_chunks():
        async for result in stream:
            print(result.record.bins)
finally:
    stream.close()

See also

limit(): Cap total records returned. has_more_chunks(): Advance to the next server chunk.

default_expire_record_after(duration)¶

Set a default TTL using a datetime.timedelta.

Equivalent to default_expire_record_after_seconds() with seconds derived from duration — applied to chained operations that lack their own TTL. A duration resolving to -1, -2, or 0 seconds selects the corresponding TTL sentinel.

Parameters:: duration (timedelta) – Time-to-live.
Return type:: Self
Returns:: self for method chaining.

default_expire_record_after_seconds(seconds)¶

Set a default TTL applied to chained operations that lack their own.

Parameters:: seconds (int) – Time-to-live in seconds. A positive value sets an explicit TTL; the sentinels -1, -2, and 0 select never-expire, no-change, and namespace-default respectively. Not range-checked here — a value the client cannot represent is rejected when the write is built.
Return type:: Self
Returns:: self for method chaining.

default_expire_record_at(when)¶

Set a default TTL so chained writes expire at an absolute point in time.

A naive when is interpreted in local time; pass a timezone-aware datetime for explicit UTC or other zones.

Parameters:: when (datetime) – Future point at which chained writes should expire.
Return type:: Self
Returns:: self for method chaining.
Raises:: ValueError – If when is not strictly in the future.

default_expiry_from_server_default()¶

Set the default TTL to the namespace’s server default (TTL = 0).

Return type:: Self

default_never_expire()¶

Set the default TTL to never expire (TTL = -1).

Return type:: Self

default_where(expression, *params)¶

Set a filter applied to any chained operation that does not call where().

Overloads:

self, expression (str), params (Any) → QueryBuilder
self, expression (FilterExpression) → QueryBuilder

When a chain contains multiple operations (reads, writes, UDFs), each operation inherits this filter unless it supplies its own where().

Example:

stream = await (
    session.upsert(k1)
        .bin("status").set_to("active")
        .where(f"$.age >= {min_age}")
    .delete(k2, k3)
    .upsert(k4)
        .bin("flag").set_to(True)
    .default_where("$.active == true")
    .execute()
)
# upsert(k1) keeps its own where(); the delete and
# second upsert inherit default_where.

Parameters:

expression (Union[str, FilterExpression]) – AEL string or FilterExpression.
*params (Any) – Values for printf placeholders in an AEL template. See where() for the interpolation contract.

Returns:

This builder for chaining.

Raises:

TypeError – If params accompany a FilterExpression.
ValueError – If the template is not a valid printf format string.

See also

where(): Per-operation filter on the current operation.

default_with_no_change_in_expiration()¶

Set the default to preserve each record’s existing TTL (TTL = -2).

Return type:: Self

delete(arg1, *more_keys)¶

Open a delete segment.

Example::: await session.delete(key).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder (often followed immediately by execute()).

effective_allow_scans_with_where()¶

Resolve whether this query may fall back to a primary-index scan.

Applies the wire precedence: the query’s hint wins when it sets allow_scans_with_where, otherwise the resolved Behavior query setting, otherwise the default (allow the fallback).

Return type:: bool
Returns:: True if a where-clause query may fall back to a primary-index (full-set) scan, False if it is rejected.

See also

QueryHint.allow_scans_with_where

execute_udf(*keys)¶

Finalize the current segment and chain a UDF execution on keys.

The current read or write segment is packaged as-is; the returned builder targets the new key(s), and the whole chain still executes as one batch when the terminal execute() runs. Call function(package, name) next.

Parameters:: *keys (Key) – One or more Key targets for the UDF segment.
Return type:: UdfFunctionBuilder
Returns:: A UdfFunctionBuilder — call function next.
Raises:: ValueError – If no keys are provided, or if the current query is a dataset (index) query, which cannot be stacked.

Example:

rs = await (
    session.query(users.id(1))
    .bin("name").get()
    .execute_udf(users.id(2))
    .function("my_module", "my_fn")
    .execute()
)

See also

execute_udf(): Start a: chain with a UDF instead.

exists(arg1, *more_keys)¶

Open an exists-check segment.

Example::: stream = await session.exists(key).execute() found = (await stream.first_or_raise()).as_bool()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder.

fail_on_filtered_out()¶

Surface reads that fail a filter as FILTERED_OUT rows instead of omitting them.

Applies to key-based reads with a where filter. Without this flag, a key whose record exists but fails the filter is absent from the stream, whether or not include_missing_keys() is set. Writes always report a filtered-out key as a FILTERED_OUT row.

Example:

stream = await (
    session.query(customers.ids(20, 21, 22))
    .where("$.name == 'Tim'")
    .fail_on_filtered_out()
    .execute()
)
async for row in stream:
    if row.result_code == ResultCode.FILTERED_OUT:
        print(f"{row.key.value} is not a Tim")

Return type:: Self
Returns:: This builder for chaining.

See also

include_missing_keys(): Rows for keys whose record does not exist.

filter(filter_obj)¶

Attach an explicit secondary-index filter to the query.

The filter is authoritative: it is sent to the server unchanged as the index access path, bypassing server-side index selection. Any where() clause on the same builder travels beside it as a residual filter expression. An index-selection hint (QueryHint.index_name) does not rewrite an explicit filter; QueryHint.query_duration still applies. A query carries at most one filter.

Example:

stream = await (
    session.query(customers)
        .filter(Filter.range_by_index("idx_customer_age", 30, 65))
        .where("$.status == 'active'")
        .execute()
)

Parameters:: filter_obj (Filter) – The secondary-index filter to attach.
Return type:: Self
Returns:: This builder for method chaining.
Raises:: ValueError – If filter has already been called on this builder.

See also

where(), with_hint()

include_missing_keys()¶

Emit a row for each requested key whose record does not exist.

Missing keys appear as non-OK RecordResult entries (KEY_NOT_FOUND_ERROR) instead of being skipped. A key whose record exists but fails a where filter is a different outcome and stays omitted; add fail_on_filtered_out() to surface those too.

Example:

stream = await (
    session.query(customers.ids(46, 47, 48))
    .include_missing_keys()
    .execute()
)
async for row in stream:
    if row.result_code == ResultCode.KEY_NOT_FOUND_ERROR:
        print(f"no customer {row.key.value}")

Return type:: Self
Returns:: This builder for chaining.

See also

fail_on_filtered_out(): Rows for keys whose record fails the filter.

insert(arg1, *more_keys)¶

Open a create-only segment (fails if the record exists).

Example::: await session.insert(key).put({“name”: “Ada”}).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder for further bins and execute().

limit(limit)¶

Cap the total number of records the query returns.

Once the limit is reached, the query stops. A query without a limit returns every matching record; there is no value that means “unlimited”, so leave the limit unset instead.

On a key query, only the first limit keys are read, in order across chained reads. A missing key still counts, so ids(1, 3, 5, 7) with limit(3) returns two records when key 5 does not exist.

Example:

stream = await session.query(users).where("$.age > 18").limit(100).execute()

Parameters:: limit (int) – Maximum number of records to return; must be positive.
Return type:: Self
Returns:: This builder for chaining.
Raises:: ValueError – If limit <= 0, or at execute if a key query chain also writes or calls a UDF.

See also

chunk_size(): Records per server round trip, rather than in total.

on_partition(part_id)¶

Target a specific partition for the query.

This method restricts the query to a single partition. This can be useful for load balancing or when you know the data distribution across partitions. On a key query, keys in other partitions are skipped before anything is sent.

Parameters:: part_id (int) – The partition ID to target (0-4095)
Return type:: Self
Returns:: self for method chaining
Raises:: ValueError – If part_id is out of range, or at execute if a key query chain also writes or calls a UDF.

Example:

query = session.query(dataset).on_partition(5)

See also

on_partition_range(): Target a contiguous span of partitions. partition(): Apply a pre-built partition filter.

on_partition_range(start_incl, end_excl)¶

Target a range of partitions for the query.

This method restricts the query to a specific range of partitions. This can be useful for load balancing, parallel processing, or when you know the data distribution across partitions.

The partition range can only be set once per query. Subsequent calls with different ranges will overwrite the previous range.

On a key query, keys outside the range are skipped before anything is sent, and limit() then counts only the keys that remain.

Parameters:

start_incl (int) – Start partition (inclusive, 0-4095)
end_excl (int) – End partition (exclusive, 1-4096)

Return type:

Self

Returns:

self for method chaining

Raises:

ValueError – If partition range is invalid, or at execute if a key query chain also writes or calls a UDF.

Example:

# Query partitions 0-2047 (first half)
query = session.query(dataset).on_partition_range(0, 2048)

# Query partitions 100-199
query = session.query(dataset).on_partition_range(100, 200)

See also

on_partition(): Target a single partition. partition(): Apply a pre-built partition filter.

partition(partition_filter)¶

Restrict the query using a PNC PartitionFilter.

Prefer on_partition() or on_partition_range() for common cases. A filter reused from an earlier dataset query resumes where that query stopped. On a key query only the filter’s partition range applies: keys outside it are skipped before anything is sent.

Example:

part = PartitionFilter.by_range(0, 2048)
stream = await session.query(users).partition(part).execute()

Parameters:: partition_filter (PartitionFilter) – Built filter (all partitions, by id, by range, etc.).
Return type:: Self
Returns:: This builder for chaining.
Raises:: ValueError – At execute, on a key query, if the filter carries resume state (PartitionFilter.by_key or progress from an earlier query), or if the chain also writes or calls a UDF.

See also

on_partition_range(): Inclusive start, exclusive end partition ids.

query(arg1, *more_keys)¶

Chain another query with new key(s) for batch/point stacking.

Finalizes the current query segment and begins a new one with the given key(s). Each segment can have its own bins, operations, and filter expression. Dataset (index) queries cannot be stacked.

Parameters:

arg1 (Union[Key, List[Key]]) – A single Key or a List[Key].
*more_keys (Key) – Additional keys (varargs).

Return type:

Self

Returns:

self for method chaining.

Raises:

ValueError – If the current query is a dataset query (no keys).

Example:

rs = await (
    session.query(users.ids(1, 2, 3))
    .bin("map").get()
    .query(users.ids(4, 5, 6))
    .bin("name").get()
    .execute()
)

records_per_second(rps)¶

Throttle the query to at most rps records per second on each server node.

Example:

stream = await session.query(users).records_per_second(1000).execute()

Parameters:: rps (int) – Maximum records per second; 0 means unthrottled.
Return type:: Self
Returns:: This builder for chaining.

See also

limit(): Cap the total records returned.

replace(arg1, *more_keys)¶

Open a full replace segment (removes bins not written in this segment).

Example::: await session.replace(key).put({“a”: 1}).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder.

replace_if_exists(arg1, *more_keys)¶

Open replace-if-exists semantics (like replace, but only if the record exists).

Example::: await session.replace_if_exists(key).put({“a”: 1}).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder.

touch(arg1, *more_keys)¶

Open a touch segment (TTL refresh).

Example::: await session.touch(key).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder.

update(arg1, *more_keys)¶

Open an update-only segment (fails if the record is missing).

Example::: await session.update(key).bin(“count”).add(1).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder for further bins and execute().

upsert(arg1, *more_keys)¶

Open a create-or-update segment for the given key(s).

Example::: await session.upsert(key).put({“name”: “Bob”}).execute()

Return type:: TypeVar(_WSB, bound= _WriteSegmentBuilderBase)
Returns:: WriteSegmentBuilder for put / bin / execute.

See also

upsert(): Session entry point.

where(expression, *params)¶

Apply a server-side filter for dataset queries or keyed reads that support it.

Overloads:

self, expression (str), params (Any) → QueryBuilder
self, expression (FilterExpression) → QueryBuilder

String arguments are AEL, compiled by the server. Pass a pre-built FilterExpression when constructing filters programmatically.

Supplying params interpolates them into the template with printf syntax. Only trusted values belong here — interpolation does not quote or escape, so it is no safer than an f-string.

Parameters:

expression (Union[str, FilterExpression]) – AEL string or FilterExpression.
*params (Any) – Values for printf placeholders in an AEL template.

Returns:

This builder for chaining.

Raises:

TypeError – If params accompany a FilterExpression.
ValueError – If the template is not a valid printf format string, or where has already been called for this operation.

Example:

qb = session.query(ds).where("$.status == 'active'")
qb = session.query(ds).where("$.score > %d", min_score)
qb = session.query(ds).where(f"$.score > {min_score}")

See also

default_where(): Default filter for chained operations without their own.

with_hint(hint)¶

Attach a query hint for secondary index selection or scheduling.

A hint can redirect which secondary index is used (index_name) or override the expected query duration (query_duration). Only one call to with_hint is allowed per builder.

Example:

stream = await (
    session.query(dataset)
        .filter(Filter.equal("age", 30))
        .with_hint(QueryHint(index_name="age_idx"))
        .execute()
)

Parameters:: hint (QueryHint) – A QueryHint instance.
Return type:: Self
Returns:: This builder for method chaining.
Raises:: ValueError – If with_hint has already been called on this builder.

See also

QueryHint

with_no_bins()¶

Specify that no bins should be read (header-only query).

This method is useful when you only need to check for record existence or get metadata like generation numbers, without reading the actual data.

This method cannot be used together with bins().

Return type:: Self
Returns:: self for method chaining.
Raises:: ValueError – If used together with bins().

with_op_projection(*ops)¶

Project query results through one or more read operations.

The server applies ops to every matching record and returns the operation results in place of the configured bin set. Mutually exclusive with bins(): when a projection is set the server ignores the bin list. Subsequent calls replace the previous projection.

The server accepts basic get_bin / get_header ops as well as CDT, expression, bit, and HLL reads — for example CdtOperation.select_values("bin", [...]).

Parameters:: *ops (Any) – One or more native aerospike_native read operations.
Return type:: Self
Returns:: This builder for method chaining.

Example:

from aerospike_sdk import CTX, CdtOperation
stream = await (
    session.query(users)
        .with_op_projection(
            CdtOperation.select_values(
                "inventory", [CTX.map_key("books")]
            ),
        )
        .execute()
)

with_txn(txn)¶

Opt this builder into (or out of) a specific transaction.

Overrides any transaction captured at construction. Pass None to opt out of an ambient transaction (useful inside a TransactionalSession when a single operation must run outside the MRT). None also opts a multi-key write out of implicit batch-write transactions (implicit_batch_write_transactions), guaranteeing the operation runs in no transaction at all.

Parameters:: txn (Optional[Txn]) – The Txn to participate in, or None to run without a transaction.
Return type:: Self
Returns:: This builder for method chaining.

Example:

async with session.transaction() as tx:
    await tx.upsert(k1).bin("v").set_to(1).execute()
    # Run this one write outside the transaction:
    await tx.upsert(k2).with_txn(None).bin("v").set_to(2).execute()