Reading Data

All reads go through session.query(), which returns a QueryBuilder. Chain methods to configure the query, then call .execute() to get a RecordStream.

Point Read (Single Key)

users = DataSet.of("test", "users")

stream = await session.query(users.id(1)).execute()
result = await stream.first_or_raise()   # takes one row, then closes the stream
print(result.record.bins)  # {'name': 'Alice', 'age': 30}

first_or_raise() is terminal — it returns the first row and closes the stream (releasing the producer), which is exactly what you want for a point read. See Taking a Single Row for the full set of single-row accessors.

Set Scan (All Records)

async with await session.query(users).execute() as stream:
    async for result in stream:
        print(result.record.bins)

Batch Read (Multiple Keys)

async with await session.query(*users.ids(1, 2, 3)).execute() as stream:
    async for result in stream:
        print(result.record.key, result.record.bins)

A multi-key read is split into per-node sub-batches, and any node whose sub-batch holds a single key is sent a regular single-record command instead of a batch request — the server executes those on the transaction service queue rather than the batch queue. The choice is per-node, so one call can mix singleton and multi-key groups. There is nothing to gain from special-casing size-1 batches in application code; the client already does.

Draining a stream fully releases it automatically; the async with above also guarantees release if you leave the loop early. See Closing a Stream.

A batch read is a single server request. When what you have instead is a window of keys you want read independently but submitted together, session.get_many sends them as separate single-record commands fused into one client-side submission and returns a result per key, positionally. It is async-only and reaches the highest single-loop read throughput — see Async window API.

Buffered vs. Lazy Delivery

Every query-path builder exposes two terminals with different result-delivery semantics. execute() is the default — reach for stream() only when a large result set makes buffering expensive.

  • execute() — buffered. Awaits every result, then returns a RecordStream backed by a fully-materialized list. Results arrive in input order, and for a chain that includes writes those writes are guaranteed complete server-side by the time the call returns. Use this for most workloads.

  • stream() — lazy. Returns a RecordStream that yields one RecordResult per __anext__ (__next__ on sync) as each node responds. The first results are available as soon as the first node responds, without waiting for the rest, so peak memory stays bounded to the in-flight node responses — useful for large batches and scans.

    Caveats:

    • Yields completion order, not input order. Each RecordResult carries its input position in .index; sort after collecting if you need positional order.

    • No writes-complete-on-return guarantee. If a chain includes writes and you discard the stream without iterating, per-node work may still be in-flight. Use execute() when you need writes done on return.

    • Per-key errors land inline on RecordResult; cluster-level errors raise mid-iteration.

    • Close it if you abandon it early (see below).

# Bounded-memory scan — process records as they arrive
async with session.query(users).stream() as stream:
    async for result in stream:
        handle(result.record)

Keyless dataset queries and scans already stream lazily from the server, so stream() and execute() deliver the same incremental behavior there; the distinction matters most for multi-key batch shapes. The same two terminals exist on the chained write builders — see Writing Data. Sync builders have the identical contract; iterate with for result in stream.

Closing a Stream

A RecordStream that is drained to exhaustion releases its underlying producer automatically. If you abandon a lazy stream early — break out of the loop, take only the first row, or hit an exception — call close() (or use the context manager) so the producer is released promptly instead of at garbage-collection time.

close():

  • stops further iteration (subsequent iteration ends immediately), and

  • releases the underlying producer: for a batch stream it drops the receiver and any buffered-but-unconsumed results; for a scan it tears down the server-side query. In-flight per-node requests still complete in the background and release their connections as they finish — close() reclaims the consumer side. It is idempotent.

The recommended pattern is the context manager, which closes on every exit path — normal completion, early break, or exception:

# async — `async with`
async with session.query(*users.ids(1, 2, 3)).stream() as stream:
    async for result in stream:
        if done(result):
            break            # close() still runs on the way out

# sync — `with`
with session.query(*users.ids(1, 2, 3)).stream() as stream:
    for result in stream:
        ...

A buffered execute() stream over a single key or a materialized list holds no external producer, so closing it is a harmless no-op — but the context-manager habit is safe to apply uniformly.

Taking a Single Row

Four accessors take a single row. They split on two independent axes — closing (does it release the stream afterward?) and error handling (return a non-OK RecordResult as data, or raise):

keeps stream open

closes stream (terminal)

returns envelope

pop()

first()

raises on error

pop_or_raise()

first_or_raise()

  • first() / first_or_raise() — take one row, then close() the stream. Use these for a point read or any “I want exactly one record” case; the producer is released immediately.

  • pop() / pop_or_raise() — take one row and leave the rest available for further iteration. Use these to peek at a head element and keep going.

  • The _or_raise variants raise on an empty stream and on a non-OK row (via RecordResult.or_raise()); the plain variants return None for empty and hand back non-OK rows as RecordResult data (is_ok=False) so you can branch on partial success.

The builder exposes first() and first_or_raise() directly, which is the shorter form for the single-record case — one await instead of two, with the same semantics, since each delegates to the stream method of the same name:

# Point read, straight from the builder.
result = await session.query(users.id(1)).first_or_raise()
record = result.record_or_raise()

# Or branch on partial success instead of raising.
result = await session.query(users.id(1)).first()
if result is None:
    ...                     # no such record
elif not result.is_ok:
    ...                     # result.result_code says why

The terminals return a RecordResult rather than a bare Record, so result_code, sub_code, and in_doubt stay available; record_or_raise() is the step that unwraps it.

# The same read through the stream, when you want the stream for other reasons.
rec = await (await session.query(users.id(1)).execute()).first_or_raise()

# Peek the head, then process the remainder — pop keeps the stream open
stream = await session.query(*users.ids(1, 2, 3)).stream()
head = await stream.pop()
async for rest in stream:
    ...
stream.close()   # or wrap the whole thing in `async with`

Sync builders and streams expose the same methods; call them without await.

Selecting Bins

Return only specific bins to reduce network transfer:

stream = await session.query(users).bins("name", "age").execute()

Or exclude all bins (metadata only):

stream = await session.query(users).with_no_bins().execute()

Projecting with Bin-Level Reads

The same .bin(...) read operations that work on a key read (get(), select_from(), CDT and string reads) also apply to a dataset query. The server evaluates them against every matching record and returns their results in place of the stored bins, so a projection can compute values the record never held:

stream = await (
    session.query(users)
    .where("$.age >= %s", 21)
    .bin("name").get()
    .bin("age_next").select_from(Exp.num_add([Exp.int_bin("age"), Exp.int_val(1)]))
    .execute()
)
async for result in stream:
    rec = result.record_or_raise()
    print(rec.bins["name"], rec.bins["age_next"])

When both are present, the read operations take precedence over .bins(...). Write, touch, and delete operations are rejected on a read query; use background_task() for those.

Filtering with AEL

Use the Aerospike Expression Language to filter records server-side:

stream = await (
    session.query(users)
    .where("$.age > 25 and $.status == 'active'")
    .execute()
)

This and the AEL queries below run fastest when a secondary index covers the filtered bin. A .where() query that no index can satisfy runs as a full-set scan unless the query or Behavior rejects that fallback — see Secondary Indexes.

Or with a pre-built FilterExpression:

from aerospike_sdk import Exp

expr = Exp.and_([
    Exp.gt(Exp.int_bin("age"), Exp.int_val(25)),
    Exp.eq(Exp.string_bin("status"), Exp.string_val("active")),
])
stream = await session.query(users).where(expr).execute()

Each query or chained operation takes one where(). A second call raises ValueError rather than replacing the first, so combine conditions in a single expression (and, or) instead.

Partition Filtering

Query a single partition:

stream = await (
    session.query(users)
    .on_partition(10)
    .execute()
)

Or a contiguous range, given an inclusive start and an exclusive end. This is how you split a scan across parallel workers — here, the first quarter of the 4096-partition space:

stream = await (
    session.query(users)
    .on_partition_range(0, 1024)
    .execute()
)

On a key query, a partition filter and limit() choose which keys are read, before anything is sent: keys outside the partition range are skipped, then only the first limit remaining keys are read. A missing key still counts toward the limit. Because skipped keys would never be written, a key chain that also writes or calls a UDF raises ValueError when either is set.

Query Policies

Fine-tune query behavior:

from aerospike_sdk import QueryDuration, QueryHint

stream = await (
    session.query(users)
    .where("$.age > 18")
    .with_hint(QueryHint(query_duration=QueryDuration.LONG))
    .records_per_second(5000)
    .chunk_size(500)
    .execute()
)

These tweaks layer onto the session’s Behavior, so its query timeouts, retries, and replica preference still apply. To change those, derive a Behavior with a reads_query=Settings(...) scope (see Connecting).

Draining a chunked query

chunk_size() sets how many records arrive per server round trip, and advancing between chunks is the caller’s job. Iterating the stream yields only the chunk already loaded, so the plain async for shown below would stop after 500 records and silently leave the rest unread. Drive has_more_chunks() in an outer loop instead:

stream = await session.query(users).where("$.age > 18").chunk_size(500).execute()
try:
    while await stream.has_more_chunks():
        async for result in stream:
            print(result.record.bins)
finally:
    stream.close()

To cap how many records you get in total — rather than how many arrive per trip — use limit().

RecordResult

Each item in the stream is a RecordResult:

async for result in stream:
    if result.is_ok:
        record = result.record
        print(record.key, record.bins, record.generation, record.expiration)
    else:
        print(f"Error: {result.to_exception()}")

Branch on is_ok rather than comparing result_code to OK: it checks both result_code and exception. A row that failed client-side, before the server answered, carries a negative client code such as ResultCode.CLIENT_ERROR (or ResultCode.TIMEOUT for a client-side deadline), with the exception holding the detail. See Error Handling.

Use record_or_raise() to raise on error results:

async for result in stream:
    record = result.record_or_raise()

Query Hints

Influence secondary index selection with QueryHint:

from aerospike_sdk import QueryHint

stream = await (
    session.query(users)
    .where("$.age > 25")
    .with_hint(QueryHint(index_name="age_idx"))
    .execute()
)