TransactionalSession¶

class aerospike_sdk.aio.transactional_session.TransactionalSession[source]¶

Bases: TransactionalSessionBase, Session

Async context manager that groups operations into a multi-record transaction.

Subclasses Session, so every session API (query, upsert, insert, batch, …) works unchanged inside async with; builders capture the active Txn via current_transaction and thread it onto every policy they hand to the PNC — the user never touches a policy.

On clean exit the transaction is committed; if an exception propagates out of the block the transaction is aborted. Explicit commit() and abort() are also available for manual control. Either finalizes the session: a later operation on it raises TransactionError with ResultCode.TXN_ALREADY_COMMITTED or ResultCode.TXN_ALREADY_ABORTED instead of running outside the transaction.

Example:

async with client.create_session().transaction() as tx:
    await tx.upsert(accounts.id("A")).bin("balance").set_to(100).execute()
    await tx.upsert(accounts.id("B")).bin("balance").set_to(200).execute()
# Auto-committed on clean exit; auto-aborted on exception.

__init__(client, behavior=None)[source]¶

Create a transactional session; prefer Session.transaction().

Parameters:

client (Client) – Connected Client.
behavior (Optional[Behavior]) – Policy bundle for operations started from this session. Defaults to Behavior.DEFAULT when omitted.

Note

Application code should not construct TransactionalSession directly; call Session.transaction() or Client.transaction() instead.

async get(key, bins=None)[source]¶

Direct single-key point read — returns Record or raises.

Bypasses the builder chain (session.query(key).execute()) and the RecordStream wrapper: one await reaches the underlying client and the resulting Record is returned unwrapped. Use when you have a single key and want minimum per-op overhead; use query() when you need filters, projections, or streaming.

Parameters:

key (Key) – Target Key.
bins (Optional[List[str]]) – Optional bin-name projection. None (default) reads all bins.

Return type:

Record

Returns:

The Record for key.

Raises:

AerospikeError – Server or client errors (including KEY_NOT_FOUND_ERROR) are raised as the SDK exception type for the failure, without being wrapped in a RecordResult.

Example:

users = DataSet.of("test", "users")
rec = await session.get(users.id(1))
name = rec.bins["name"]

See also

query(): Builder-based reads for projections, streams, and secondary-index queries. put(): Direct single-key upsert.

async put(key, bins)[source]¶

Direct single-key upsert — returns None or raises.

Bypasses the builder chain (session.upsert(key).put(...).execute()) and the RecordStream wrapper: one await reaches the underlying client. Use when you have a single key and want minimum per-op overhead; use upsert() when you need atomic multi-op semantics, TTL overrides, generation checks, durable delete, or filter expressions.

Parameters:

key (Key) – Target Key.
bins (Dict[str, Any]) – Mapping of bin name to value to write. An empty mapping is permitted.

Return type:

None

Returns:

None on success.

Raises:

AerospikeError – Server or client errors are raised as the SDK exception type for the failure.

Example:

users = DataSet.of("test", "users")
await session.put(users.id(1), {"name": "Tim", "age": 30})

See also

upsert(): Builder-based writes with full feature set. get(): Direct single-key point read.

async get_many(keys, bins=None)[source]¶

Direct point reads for a window of keys — one await, one result list.

Client-side fusion of independent single-record reads: one call submits the whole window and one completion delivers all results, so per-op submission and wakeup overhead is amortized across the window. These are NOT a server batch request — each key remains an independent wire op (use query() with a key list for wire batching).

Parameters:

keys (List[Key]) – Target keys; results are positional.
bins (Optional[List[str]]) – Optional bin-name projection shared by the window.

Return type:

List[Union[Record, AerospikeError]]

Returns:

A list the same length as keys. Each slot is the Record for that key, or the AerospikeError instance (not raised) for that key — check with isinstance(slot, AerospikeError). One failed key never fails its window-mates.

Records are handed back raw rather than wrapped in RecordResult: a wrapper per slot would cost one allocation per key, which is the per-op cost this API exists to amortize. The asymmetry with the builder surface is deliberate.

Raises:

AerospikeError – When the whole-window submission fails (for example, the client was closed); per-key failures land in result slots instead.

Example:

users = DataSet.of("test", "users")
records = await session.get_many([users.id(i) for i in range(16)])
found = [r for r in records if not isinstance(r, AerospikeError)]

See also

get(): Single-key point read. put_many(): Window counterpart for writes.

async put_many(keys, bins)[source]¶

Direct upserts of one payload to a window of keys — one await.

Write counterpart of get_many(): the bins payload is converted once and written to every key in the window as independent wire ops with client-side fused submission/completion.

Parameters:

keys (List[Key]) – Target keys; results are positional.
bins (Dict[str, Any]) – Mapping of bin name to value written to every key.

Returns:

None for each success, or the AerospikeError instance (not raised) for that key. Slots are bare rather than RecordResult for the same allocation reason as get_many().

Return type:

List[Optional[AerospikeError]]

Raises:

AerospikeError – When the whole-window submission fails; per-key failures land in result slots instead.

Example:

users = DataSet.of("test", "users")
outcomes = await session.put_many(
    [users.id(i) for i in range(16)], {"active": True})
errors = [e for e in outcomes if e is not None]

See also

put(): Single-key upsert. get_many(): Window counterpart for reads.

async do_in_transaction(operation, *, max_attempts=None, sleep_between_retries=None)[source]¶

Join this transaction instead of starting a nested one.

A transaction cannot contain another transaction, so the useful reading of a nested call is “run this as part of the transaction I am already in”. Opening a second one instead would silently split the caller’s work across two transactions that commit independently – losing the atomicity the outer call was written to get.

The outermost call owns the commit and the retrying, so the retry arguments do not apply here and are accepted only to keep the signature substitutable.

Parameters:

operation (Callable[[TransactionalSession], Awaitable[Any]]) – Async callable receiving this session.
max_attempts (Optional[int]) – Ignored; the outermost call owns retrying.
sleep_between_retries (Optional[float]) – Ignored, for the same reason.

Return type:

Any

Returns:

Whatever operation returns.

Raises:

RuntimeError – If this session’s transaction is no longer active.

Example:

async def transfer(tx):
    await tx.upsert(src).put({"balance": 90}).execute()
    # Joins the transaction already in progress.
    return await tx.do_in_transaction(audit)

await session.do_in_transaction(transfer)

See also

aerospike_sdk.aio.session.Session.do_in_transaction(): The: outermost entry point, which does open a transaction.

async commit()[source]¶

Commit the transaction and return the server-reported status.

ROLL_FORWARD_CLOSE_ABANDONED is still success: the writes are durable and only monitor cleanup was left to the server. An abandoned roll-forward raises instead — those writes are not yet visible.

A commit that fails in doubt — the roll-forward mark may or may not have reached the server — leaves the transaction open and retryable, and calling this again is the only way to resolve it: abort() is refused from that point, because rolling back could discard writes the server is committing. Every other commit failure finalizes the session as before.

Raises:

RuntimeError – If the transaction was never started.
TransactionError – With ResultCode.TXN_ALREADY_ABORTED if the transaction was aborted.
CommitError – The roll-forward was abandoned, or another commit stage failed.

Return type:

TxnStatus

Returns:

COMMITTED, ALREADY_COMMITTED on a repeat call, or another TxnStatus the server reported.

See also

abort(): Undo the transaction instead of committing.

async abort()[source]¶

Abort the transaction and return the server-reported status.

Refused after a commit failed in doubt: the server may be rolling the transaction forward, and rolling back would discard those writes. The transaction stays open in that case, so commit() can still be retried.

Raises:

RuntimeError – If the transaction was never started.
TransactionError – With ResultCode.TXN_ALREADY_COMMITTED if the transaction was committed.
AerospikeError – If the abort was refused because a commit already failed in doubt, or the roll-back itself failed.

Return type:

TxnStatus

Returns:

ABORTED, ALREADY_ABORTED on a repeat call, or another TxnStatus the server reported.

See also

commit(): Persist the transaction instead of aborting.

property active: bool¶

True when a transaction has been started and not yet finalized.

Returns:: Whether a transaction is currently active on this session.

Example:

with session.transaction() as tx:
    assert tx.active

execute_udf(*keys)¶

Run a registered server-side UDF on one or more keys (foreground).

Chain function(package, name) (package is the registered module name without .lua), optional passing(*args) for Lua parameters, optional where for a filter expression, then await ...execute() to obtain a RecordStream. Multiple keys use a batch UDF; results preserve per-key order where applicable.

Parameters:

*keys (Key) – One or more Key targets in the same namespace and set.

Return type:

Any

Returns:

UdfFunctionBuilder — call function next.

Raises:

ValueError – If no keys are given.
RuntimeError – If the client is not connected.

Example:

users = DataSet.of("test", "users")
stream = await (
    session.execute_udf(users.id("a"))
    .function("my_module", "my_fn")
    .passing("binName", 42)
    .execute()
)
value = await stream.first_udf_result()

See also

query(): Read bins without UDF. background_task(): Dataset-scoped background UDF.

property txn: Txn¶

Return the active Txn.

Raises:: RuntimeError – If the session has not been entered (no active txn).
Returns:: The active Txn.

Example:

with session.transaction() as tx:
    assert tx.txn is not None