RecordStream

class aerospike_sdk.record_stream.RecordStream

Bases: object

Async iterator of RecordResult rows.

Produced by await session.query(...).execute() / .stream() and similar APIs — you receive a stream from an operation rather than constructing one. Prefer async for row in stream, or helpers such as collect() and first().

Example:

Typical consumption with ``async for``::

    stream = await session.query(key).bins(["name"]).execute()
    async for row in stream:
        if row.is_ok and row.record:
            print(row.record.bins)

See also

first_or_raise(): Assert a single OK row.

__init__(source)

classmethod chain(streams)

Yield all results from each stream in order.

Return type:

RecordStream

Example::

combined = RecordStream.chain([stream_a, stream_b])

async has_more_chunks()

Check whether more server-side chunks remain.

On the first call this returns True so the caller enters the iteration loop for the already-loaded first chunk. Subsequent calls inspect the server’s PartitionFilter cursor: if more partitions remain, a new query round-trip is issued transparently and True is returned.

Returns False when: * the server cursor is done (all partitions scanned), or * the overall limit has been reached, or * the stream was not created with _from_chunked_pnc_recordset().

Raises:

AerospikeError – The cursor read or the next chunk’s query failed.

Example:

stream = await session.query(SET).chunk_size(10).execute()
chunk = 0
while await stream.has_more_chunks():
    chunk += 1
    print(f"Chunk: {chunk}")
    async for rr in stream:
        print(rr.record.bins)

Return type:

bool

async pop()

Advance one row and return it (or None if empty), keeping the stream open.

This is the non-closing counterpart of first(): it consumes a single row and leaves the rest available for further iteration (async for, collect(), or another pop()). Per-record failures come back as data on the RecordResult (is_ok=False); cluster-level errors raise from iteration.

Return type:

RecordResult | None

Returns:

The next RecordResult, or None when the stream is exhausted.

Example:

stream = await session.query(*users.ids(1, 2, 3)).execute()
head = await stream.pop()          # first row; stream stays open
rest = await stream.collect()      # the remaining rows

See also

first(): Same, but closes the stream afterward (terminal). pop_or_raise(): Raise instead of returning an error envelope.

async pop_or_raise()

Advance one row and require success, keeping the stream open.

Non-closing counterpart of first_or_raise().

Return type:

RecordResult

Returns:

The next OK RecordResult.

Raises:

• RecordNotFoundError – If the stream is exhausted (empty).

• AerospikeError – If the row is not OK (from RecordResult.or_raise()).

See also

first_or_raise(): Terminal variant that closes afterward.

async first()

Return the first row (or None if empty), then close the stream.

Terminal convenience for the common “I want a single record” case: it takes one row via pop() and then close()s the stream, releasing the underlying producer. Per-record failures come back as data (is_ok=False); cluster-level errors raise.

Use pop() instead if you intend to keep iterating.

Return type:

RecordResult | None

Returns:

The first RecordResult, or None when the stream is empty.

Example:

row = await (await session.query(key).execute()).first()
if row is None:
    ...

See also

pop(): Non-closing counterpart. first_or_raise(): Raise instead of returning an error envelope. first(): Same terminal on the builder, one await.

async first_or_raise()

Return the first row, require success, then close the stream.

Terminal counterpart of pop_or_raise().

Return type:

RecordResult

Returns:

The first OK RecordResult.

Raises:

• RecordNotFoundError – If the stream yields no rows (empty).

• AerospikeError – If the first row is not OK (from RecordResult.or_raise()).

Example:

rec = (await stream.first_or_raise()).record_or_raise()

See also

pop_or_raise(): Non-closing counterpart. first_or_raise(): Same terminal on the builder.

async first_udf_result()

Scan forward for the first non-None udf_result.

Return type:

Any | None

Returns:

The UDF return value, or None if no row carries a UDF result.

Example::

value = await stream.first_udf_result()

See also

Session.execute_udf(): Produces streams with UDF results.

async collect()

Drain the stream into a list (order preserved).

Return type:

list[RecordResult]

Returns:

All remaining RecordResult instances.

Example:

rows = await stream.collect()
oks = [r for r in rows if r.is_ok]

async failures()

Drain the stream and return rows where is_ok is false.

Return type:

list[RecordResult]

Returns:

Only error or non-OK rows.

Note

Like collect(), this consumes the entire stream.

close()

Stop iteration and release the underlying producer.

Marks the stream closed so any further async for / __anext__() ends immediately, and — for lazily-fed streams (a batch stream or a query recordset) — forwards to the underlying producer’s close() so its receiver and any buffered-but-unconsumed results are released now, rather than at garbage-collection time.

For a batch stream, in-flight per-node requests still complete in the background and release their connections as they finish; close() reclaims the consumer side. For a query recordset, the server-side scan is torn down.

Idempotent. Always call this when abandoning a lazy stream before it is fully drained; a fully-drained stream releases automatically.

Return type:

None