Writing Data¶

All writes go through session entry points that return a WriteSegmentBuilder. Assign bin values via dict (or chained bin operations), then call .execute().

Write Verbs¶

Method	Behavior	Record must exist?
`upsert()`	Create or update	No
`insert()`	Create only	No (fails if exists)
`update()`	Update only	Yes (fails if missing)
`replace()`	Replace all bins	Yes (fails if missing)
`delete()`	Remove record	No
`touch()`	Reset TTL	Yes
`exists()`	Check existence	No

Dict Pattern¶

Set bins by passing a dict to .put() — the common case:

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

await (
    session.upsert(users.id(1))
    .put({"name": "Alice", "age": 30, "active": True})
    .execute()
)

Bin Chaining¶

Set bins individually with the chainable .bin().set_to() pattern — useful when building bins conditionally or mixing in per-bin operations:

await (
    session.upsert(users.id(1))
    .bin("name").set_to("Alice")
    .bin("age").set_to(30)
    .bin("active").set_to(True)
    .execute()
)

Increment¶

await (
    session.update(users.id(1))
    .bin("login_count").add(1)
    .execute()
)

GeoJSON Bins¶

Use set_to_geo_json(...) to write a bin as a GeoJSON value. The bin’s server-side particle type is GEOJSON, not STRING, which makes it eligible for GEO2DSPHERE indexing and geoCompare(...) queries.

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

await (
    session.upsert(places.id("space_needle"))
    .bin("loc").set_to_geo_json('{"type":"Point","coordinates":[-122.349,47.620]}')
    .execute()
)

AeroCircle and Polygon values use the same method — only the GeoJSON string differs.

HyperLogLog Bins¶

HllConfig describes a sketch’s precision (index_bit_count + optional min_hash_bit_count). Initialize a new sketch with hll_init(...), then accumulate elements with hll_add(...):

from aerospike_sdk import HllConfig

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

await (
    session.upsert(visitors.id("day_1"))
    .bin("h").hll_init(HllConfig.of(14))
    .bin("h").hll_add(["user-1", "user-2", "user-3"])
    .execute()
)

Each write method (hll_init, hll_add, hll_set_union) accepts four keyword-only flags:

Flag	Effect
`create_only`	Fail with `BIN_EXISTS_ERROR` if the bin already exists. Mutually exclusive with `update_only`.
`update_only`	Fail with `BIN_NOT_FOUND` if the bin does not already exist. Mutually exclusive with `create_only`.
`no_fail`	Suppress the mode-constraint error from `create_only` / `update_only` and silently no-op instead of raising. Has no effect on other server-side errors.
`allow_fold`	(`hll_set_union` only) Allow union sources at differing precisions — the server folds higher-precision inputs down to the target’s `index_bit_count`. Rejected with `PARAMETER_ERROR` on `hll_init`.

Passing both create_only=True and update_only=True raises ValueError immediately at the call site (before the wire request).

# Create the sketch only if it doesn't exist; don't error if it does.
await (
    session.upsert(visitors.id("day_2"))
    .bin("h").hll_init(HllConfig.of(14), create_only=True, no_fail=True)
    .execute()
)

To inspect a sketch’s bit widths, call hll_describe() and decode the two-element list result via RecordResult.get_hll_config(bin_name):

rs = await session.query(visitors.id("day_1")).bin("h").hll_describe().execute()
result = await rs.first_or_raise()
config = result.get_hll_config("h")
# config == HllConfig.of(14)

Insert (Fail if Exists)¶

stream = await (
    session.insert(users.id(99))
    .put({"name": "New User"})
    .execute()
)
result = await stream.first_or_raise()

Replace (Overwrite All Bins)¶

await (
    session.replace(users.id(1))
    .put({"name": "Alice Updated"})
    .execute()
)
# Only "name" bin remains; "age" and "active" are removed

Delete¶

# Single key
await session.delete(users.id(1)).execute()

# Multiple keys
await session.delete(*users.ids(1, 2, 3)).execute()

Durable delete¶

Aerospike supports two delete modes:

a normal delete that removes the record (and its lineage) outright, and
a durable delete that leaves a tombstone so a strongly-consistent (SC) cluster can resolve the deletion across partitions.

The SDK exposes both per-operation overrides and builder defaults:

Method	Scope	Effect
`with_durable_delete()`	one operation	Force durable delete for this delete only
`without_durable_delete()`	one operation	Force a non-durable delete for this delete only
`default_with_durable_delete()`	builder	Prefer durable when resolving Behavior defaults — typical for SC namespaces
`default_without_durable_delete()`	builder	Prefer non-durable when resolving Behavior defaults

The override (with_* / without_*) wins over the default; the default folds into Behavior settings resolution.

# Force durable on this single delete (per-op override)
await session.delete(users.id(5)).with_durable_delete().execute()

# Use durable as the default for every delete in this segment (SC-friendly)
await (
    session.delete(*users.ids(1, 2, 3))
    .default_with_durable_delete()
    .execute()
)

Conditional Writes¶

Filter writes server-side with .where():

await (
    session.update(users.id(1))
    .where("$.age >= 18")
    .bin("verified").set_to(True)
    .execute()
)

Records that don’t match the filter are skipped. Use .fail_on_filtered_out() to raise an error instead:

await (
    session.update(users.id(1))
    .where("$.age >= 18")
    .fail_on_filtered_out()
    .bin("verified").set_to(True)
    .execute()
)

Generation Check (Optimistic Locking)¶

await (
    session.update(users.id(1))
    .ensure_generation_is(5)
    .bin("balance").set_to(100)
    .execute()
)

TTL / Expiration¶

Three equivalent shapes for setting a record’s TTL — pick whichever your call site already has handy:

from datetime import datetime, timedelta, timezone

# Raw seconds:
await session.upsert(users.id(1)).expire_record_after_seconds(3600).put({"session_token": "abc123"}).execute()

# A timedelta:
await session.upsert(users.id(1)).expire_record_after(timedelta(hours=1)).put({"session_token": "abc123"}).execute()

# An absolute point in time:
await session.upsert(users.id(1)).expire_record_at(datetime.now(timezone.utc) + timedelta(hours=1)).put({"session_token": "abc123"}).execute()

expire_record_at accepts either a timezone-aware or naive datetime (a naive value is interpreted in local time) and raises ValueError if the datetime is in the past.

The expire_record_after_seconds / expire_record_after forms are pass-through — they do not range-check the value. Non-positive values select the TTL sentinels:

Value	Meaning	Named equivalent
`-1`	never expire	`never_expire()`
`-2`	keep the existing TTL	`with_no_change_in_expiration()`
`0`	use the namespace default	`expiry_from_server_default()`

So expire_record_after_seconds(-1) is equivalent to never_expire(). Only a value the client cannot represent — a negative other than the sentinels, or one outside the 32-bit range — is rejected, and only when the write is built.

Chain-wide defaults¶

Each verb above has a default_ counterpart that applies to every operation in the chain that does not set its own TTL — default_expire_record_after, default_expire_record_after_seconds, default_expire_record_at, default_never_expire, default_with_no_change_in_expiration, and default_expiry_from_server_default. default_where does the same for filters.

They are available from any segment, so a chain opened with a write verb can set them directly:

await (
    session.upsert(users.id(1))
    .default_expire_record_after(timedelta(minutes=10))
    .bin("session_token").set_to("abc123")
    .execute()
)

A default is only a default: an operation that sets its own TTL keeps it.

await (
    session.upsert(users.id(1))
    .default_expire_record_after(timedelta(minutes=10))
    .expire_record_after_seconds(120)      # this operation expires in 2 minutes
    .bin("session_token").set_to("abc123")
    .execute()
)

The scope is the chain, and it ends at execute() — a default never carries over to the next chain, even from the same session. Within the chain it applies to every operation regardless of where you set it (before or after the operations it covers), and to every key of a multi-key segment:

await (
    session.upsert(*users.ids(1, 2, 3))
    .default_expire_record_after(timedelta(minutes=10))
    .bin("status").set_to("migrated")
    .execute()
)                                          # all three records expire in 10 minutes

await session.upsert(users.id(4)).bin("status").set_to("migrated").execute()
                                           # new chain: no TTL

For a TTL on every write an application makes, set it on the Behavior the session carries instead — that is session-scoped. A chain default is for one unit of work.

Batch Writes¶

Multiple keys with the same operation:

await (
    session.upsert(*users.ids(1, 2, 3))
    .bin("status").set_to("migrated")
    .execute()
)

Mixed operations across different keys are handled automatically when you chain multiple write segments.

As with batch reads, any node whose sub-batch holds a single key is sent a regular single-record command instead of a batch request, automatically and per-node — there is nothing to gain from splitting out size-1 groups in application code.

A UDF apply can be part of that mix. Chain execute_udf(*keys).function(...) into a batch alongside reads, writes, and deletes; the per-key UDF applies travel in the same single round-trip rather than a separate call:

await (
    session.query(users.id(1))                       # read
    .execute_udf(users.id(2))                        # per-key UDF apply
        .function("stats", "increment").passing("visits", 1)
    .delete(users.id(3))                             # delete
    .execute()
)

A batch that is only UDF applies (no other op types) uses the dedicated apply-many path instead — both produce one round-trip.

As with batch reads, a batch write is a single server request. To write one payload to a window of keys as independent commands submitted together, session.put_many(keys, bins) fuses them into one client-side submission and reports an outcome per key, positionally. It is async-only — see Async window API.

Tabular Writes¶

When many records share the same bins, declare the bins once and supply one row per record. The verb takes the dataset rather than keys; each row starts with the record’s id and continues with one value per bin, in order:

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

await (
    session.upsert(users).bins("name", "age")
    .row(1, "Tim", 312)
    .row(2, "Bob", 25)
    .row(3, "Jane", 46)
    .execute()
)

rows(...) takes any iterable of (id, *values) sequences, so data you already hold as tuples, or a generator over a file, goes straight in:

people = [(10, "Tim", 312), (11, "Bob", 25), (12, "Jane", 46)]
await session.upsert(users).bins("name", "age").rows(people).execute()

The rows leave as one batch and come back as one RecordStream with a result per row, exactly as a hand-chained upsert(key).put(...) sequence would. insert, update and replace take the same shape and report their usual per-record outcomes, so an insert row whose record already exists shows up as KEY_EXISTS_ERROR on that row alone.

The TTL and generation verbs apply to the most recent row; the default_ verbs cover every row that sets none of its own, and with_txn runs the whole batch in a transaction:

await (
    session.insert(users).bins("name", "age")
    .row(1, "Tim", 312)
    .row(2, "Bob", 25).expire_record_after(timedelta(hours=1))
    .row(3, "Jane", 46).ensure_generation_is(4)
    .default_expire_record_after(timedelta(days=30))
    .execute()
)

A row with the wrong number of values is rejected when it is added, not on the server.

`execute()` vs `stream()`¶

Every query-path builder — the session.query() chain and the chained write segments — exposes two terminal methods with different result-delivery semantics. execute() is the default; reach for stream() only when buffering a large result set is expensive.

execute() — buffered. Awaits every per-key result, then returns a RecordStream backed by a fully-materialized list. Writes are guaranteed to have completed server-side by the time this method returns; subsequent reads observe the new state without races. Safe for “fire-and-forget” use (await the call and discard the returned stream). 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. Peak memory stays bounded to the in-flight node responses — useful for large batches where buffering the full result list would be expensive.

Caveats for stream():
- Yields completion order, not input order. Each RecordResult carries its originating op’s input position in .index. Sort after collecting if you need positional results.
- No writes-complete-on-return guarantee. If you discard the returned stream without iterating, per-node tasks may still be in-flight when subsequent code runs — subsequent reads can race against pending writes.
- Per-key errors land inline on RecordResult (.is_ok=False, .exception); cluster-level errors raise mid-iteration.
- Close it if you abandon it early — drain fully, or use the context manager, so the producer is released promptly. See Closing a Stream.

# Lazy streaming — process records as they arrive.
# `async with` releases the stream on every exit path.
async with (
    session.upsert(key1).bin("v").set_to(1)
    .upsert(key2).bin("v").set_to(2)
    .stream()
) as stream:
    async for result in stream:
        print(result.index, result.is_ok, result.key.value)

# When positional ordering is needed, collect + sort
async with (...).stream() as stream:
    results = await stream.collect()
results.sort(key=lambda r: r.index)

Sync siblings: the sync chained builders’ execute() and .stream() have the same contract; iterate the latter with for result in stream (rather than async for) and use a plain with block to auto-close.