# Copyright 2025-2026 Aerospike, Inc.
#
# Portions may be licensed to Aerospike, Inc. under one or more contributor
# license agreements WHICH ARE COMPATIBLE WITH THE APACHE LICENSE, VERSION 2.0.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Synchronous SDK session.
IO methods call PNC's ``_blocking`` entries; builder factories return
the synchronous builders (:class:`QueryBuilder`, etc.).
"""
from __future__ import annotations
import time
import typing
from datetime import timedelta
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union, overload
from aerospike_native import Key, Record, Txn, UDFLang
from aerospike_sdk.aio.background import BackgroundTaskSession as AsyncBackgroundTaskSession
from aerospike_sdk.txn_shared import (
is_retryable_txn_error,
resolve_retry_plan,
retry_delay_seconds,
)
from aerospike_sdk.dataset import DataSet
from aerospike_sdk.info_types import NamespaceDetail
from aerospike_sdk.exceptions import (
AerospikeError,
PncAerospikeError,
PncServerError,
_convert_pnc_exception,
)
from aerospike_sdk.session_shared import (
NamespaceScStatus,
SessionBase,
_namespace_sc_status,
)
from aerospike_sdk.policy.behavior import Behavior, OpKind, OpShape
from aerospike_sdk.policy.behavior_settings import Mode
from aerospike_sdk.policy.policy_mapper import to_read_policy, to_write_policy
from aerospike_sdk.metrics import usage
from aerospike_sdk.sync.background import BackgroundTaskSession
from aerospike_sdk.sync.info import InfoCommands
from aerospike_sdk.sync.operations.index import IndexBuilder
from aerospike_sdk.sync.operations.query import (
DataSetWriteBuilder, QueryBuilder, WriteSegmentBuilder, _SingleKeyWriteSegment,
)
from aerospike_sdk.sync.operations.udf import UdfFunctionBuilder
if TYPE_CHECKING:
from aerospike_native import (
AdminPolicy,
CollectionIndexType,
CTX,
DropIndexTask,
FilterExpression,
IndexTask,
IndexType,
RegisterTask,
UdfRemoveTask,
)
from aerospike_sdk.sync.client import SyncClient
from aerospike_sdk.sync.transactional_session import TransactionalSession
[docs]
class Session(
SessionBase[WriteSegmentBuilder, QueryBuilder, "TransactionalSession", DataSetWriteBuilder],
):
"""Run session-scoped reads and writes without ``async``/``await``.
Construct via :meth:`Cluster.create_session
<aerospike_sdk.sync.cluster.Cluster.create_session>`, not directly.
See Also:
:class:`~aerospike_sdk.aio.session.Session`: Async equivalent.
"""
[docs]
def __init__(
self, client: SyncClient, behavior: Behavior,
) -> None:
"""Attach a client and behavior; prefer :meth:`SyncClient.create_session`."""
self._client = client
self._behavior = behavior
# Pre-compute base policies once per session so the fast-path
# get/put + builder bypasses skip the policy_mapper for the common
# no-override case. Cache both AP and SC variants so bypass paths
# can pick the right policy per resolved namespace mode without
# rebuilding. `_cached_*_policy` stays as the AP alias.
self._refresh_cached_policies()
# Config hot-reload pushes rebuilt policies into live sessions
# (weak registration; no per-operation check).
behavior._register_session(self)
# Cache the PNC client for fast-path methods.
self._pnc_client = client.underlying_client
# Non-transactional sessions always return None;
# TransactionalSession overrides this to yield its active Txn.
self._txn: Optional[Txn] = None
[docs]
def session_for(self, behavior: Behavior) -> "Session":
"""A new session on the same cluster with a different behavior.
Shorthand for ``cluster.create_session(behavior)``; this session is
left as it is. Called on a transactional session it returns a plain
session, outside the transaction.
Args:
behavior: The behavior for the new session.
Returns:
A new :class:`Session`.
Example::
reads = session.session_for(Behavior.READ_FAST)
row = reads.query(key).execute().first_or_raise()
"""
return self._client.create_session(behavior)
# -- State accessors ------------------------------------------------------
def _resolve_namespace_mode_blocking(self, namespace: str) -> Mode:
"""Resolve AP vs SC for ``namespace`` synchronously (delegates to client)."""
return self._client._resolve_namespace_mode_blocking(namespace)
# -- Direct single-key fast paths -----------------------------------------
[docs]
def get(
self, key: Key, bins: Optional[List[str]] = None,
) -> Record:
"""Direct single-key point read — returns ``Record`` or raises.
Bypasses the builder chain (``session.query(key).execute()``) and
the :class:`~aerospike_sdk.sync.record_stream.RecordStream` wrapper:
one blocking call reaches the underlying client and the resulting
:class:`~aerospike_native.Record` is returned unwrapped. Passes the
AP + SC cached policies; PNC picks the right one based on the key's
namespace mode (from the in-memory partition map).
Args:
key: Target :class:`~aerospike_native.Key`.
bins: Optional bin-name projection. ``None`` (default) reads
all bins.
Returns:
The :class:`~aerospike_native.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
:class:`~aerospike_sdk.record_result.RecordResult`.
Example::
users = DataSet.of("test", "users")
rec = session.get(users.id(1))
name = rec.bins["name"]
See Also:
:meth:`query`: Builder-based reads for projections, streams, and secondary-index queries.
:meth:`put`: Direct single-key upsert.
"""
try:
if self._txn is None:
return self._pnc_client.get_blocking(
key, bins,
policy=self._cached_read_policy,
policy_sc=self._cached_read_policy_sc,
)
# Under MRT the cached policies are skipped (txn not stamped);
# rebuild a per-call policy from behavior.
policy = to_read_policy(
self._behavior.get_settings(OpKind.READ, OpShape.POINT))
policy.txn = self._txn
return self._pnc_client.get_blocking(key, bins, policy=policy)
except (PncServerError, PncAerospikeError) as e:
raise _convert_pnc_exception(e) from e
[docs]
def put(self, key: Key, bins: Dict[str, Any]) -> None:
"""Direct single-key upsert — no builder, no stream — synchronous.
Passes the AP + SC cached policies; PNC picks the right one based
on the key's namespace mode.
Raises:
AerospikeError: Server or client errors are raised as the SDK
exception type for the failure.
"""
try:
if self._txn is None:
self._pnc_client.put_blocking(
key, bins,
policy=self._cached_write_policy,
policy_sc=self._cached_write_policy_sc,
)
return
policy = to_write_policy(
self._behavior.get_settings(
OpKind.WRITE_NON_RETRYABLE, OpShape.POINT))
policy.txn = self._txn
self._pnc_client.put_blocking(key, bins, policy=policy)
except (PncServerError, PncAerospikeError) as e:
raise _convert_pnc_exception(e) from e
[docs]
def truncate(self, dataset: DataSet, before_nanos: Optional[int] = None) -> None:
"""Truncate a set, synchronously (PNC ``truncate_blocking``).
.. warning::
A ``dataset`` with no set truncates the **entire namespace**. See
:meth:`~aerospike_sdk.aio.session.Session.truncate`.
"""
client = self._client
if client._record_on:
usage.record_call(client, (usage.ADMIN_TRUNCATE,))
self._pnc_client.truncate_blocking(
dataset.namespace, dataset.set_name, before_nanos,
)
# -- Info / namespace SC --------------------------------------------------
[docs]
def namespace_sc_status(self, namespace: str) -> NamespaceScStatus:
"""Describe whether a namespace is SC; includes a reason when it is not."""
try:
result = self._pnc_client.info_blocking(f"namespace/{namespace}")
except Exception as e:
raise ValueError(f"Failed to check namespace '{namespace}': {e}") from e
return _namespace_sc_status(
namespace, NamespaceDetail.from_response(result, namespace)
)
[docs]
def is_namespace_sc(self, namespace: str) -> bool:
"""``True`` if ``namespace`` is in strong-consistency mode."""
return self.namespace_sc_status(namespace).is_sc
@overload
def info(self) -> InfoCommands: ...
@overload
def info(self, command: str) -> Dict[str, str]: ...
[docs]
def info(
self, command: Optional[str] = None,
) -> Union[InfoCommands, Dict[str, str]]:
"""Sync info: return :class:`~aerospike_sdk.sync.info.InfoCommands` or raw blocking result."""
if command is not None:
return self._pnc_client.info_blocking(command)
return InfoCommands(self._pnc_client)
# -- Builder factories ----------------------------------------------------
# ``query`` is inherited from SessionBase (shared arg normalization); only
# the tree-specific builder construction lives here, behind the hooks the
# base routes to. Both delegate to :meth:`_build_sync_query_builder`, which
# is also reused by :meth:`execute_udf`.
def _fast_query_builder(self, key: Key) -> QueryBuilder:
"""Single-key query builder (bench-hot ``session.query(key)`` shape)."""
return self._build_sync_query_builder(dataset=None, key=key, keys=None)
def _build_query_builder(
self, *, dataset: Optional[DataSet], keys: Optional[List[Key]],
) -> QueryBuilder:
"""Dataset / multi-key query builder (non-single-key shapes)."""
return self._build_sync_query_builder(dataset=dataset, key=None, keys=keys)
def _build_sync_query_builder(
self,
*,
dataset: Optional[DataSet],
key: Optional[Key],
keys: Optional[List[Key]],
) -> QueryBuilder:
"""Construct a :class:`QueryBuilder` with full session context.
Returns the builder pre-populated with behavior, indexes monitor,
cached policies, txn, and namespace-mode resolver.
"""
if key is not None:
builder = QueryBuilder(
client=self._pnc_client,
namespace=key.namespace,
set_name=key.set_name,
behavior=self._behavior,
cached_read_policy=self._cached_read_policy,
cached_write_policy=self._cached_write_policy,
cached_read_policy_sc=self._cached_read_policy_sc,
cached_write_policy_sc=self._cached_write_policy_sc,
cached_read_operate_policy=self._cached_read_operate_policy,
cached_read_operate_policy_sc=self._cached_read_operate_policy_sc,
txn=self._txn,
namespace_mode_resolver=None,
namespace_mode_resolver_blocking=self._resolve_namespace_mode_blocking,
sdk_client=self._client,
)
builder._single_key = key
return builder
if keys is not None:
ns = keys[0].namespace
sn = keys[0].set_name
builder = QueryBuilder(
client=self._pnc_client,
namespace=ns,
set_name=sn,
behavior=self._behavior,
cached_read_policy=self._cached_read_policy,
cached_write_policy=self._cached_write_policy,
cached_read_policy_sc=self._cached_read_policy_sc,
cached_write_policy_sc=self._cached_write_policy_sc,
txn=self._txn,
namespace_mode_resolver=None,
namespace_mode_resolver_blocking=self._resolve_namespace_mode_blocking,
sdk_client=self._client,
)
builder._keys = keys
return builder
assert dataset is not None
return QueryBuilder(
client=self._pnc_client,
namespace=dataset.namespace,
set_name=dataset.set_name,
behavior=self._behavior,
cached_read_policy=self._cached_read_policy,
cached_write_policy=self._cached_write_policy,
cached_read_policy_sc=self._cached_read_policy_sc,
cached_write_policy_sc=self._cached_write_policy_sc,
txn=self._txn,
namespace_mode_resolver=None,
namespace_mode_resolver_blocking=self._resolve_namespace_mode_blocking,
sdk_client=self._client,
)
[docs]
def background_task(self) -> BackgroundTaskSession:
"""Start a background dataset task chain (synchronous)."""
# BackgroundTaskSession needs a session-like parent for behavior etc.
# The aio variant accepts our sync session via duck typing; if not,
# we'd need a thin proxy. The aio constructor only reads state, no IO.
inner = AsyncBackgroundTaskSession(self) # type: ignore[arg-type]
return BackgroundTaskSession(inner)
[docs]
def execute_udf(self, *keys: Key) -> UdfFunctionBuilder:
"""Begin a foreground UDF invocation (synchronous)."""
if not keys:
raise ValueError("execute_udf requires at least one key")
builder = self._build_sync_query_builder(
dataset=None, key=keys[0] if len(keys) == 1 else None,
keys=list(keys) if len(keys) > 1 else None,
)
self._bind_txn(builder)
builder._op_type = "execute_udf"
return UdfFunctionBuilder(builder)
[docs]
def index(self, dataset: DataSet, /) -> IndexBuilder:
"""Synchronous secondary-index builder for a dataset."""
return self._client.index(dataset)
def _txn_session_cls(self) -> "type[TransactionalSession]":
"""Return the sync transactional-session class (late import breaks the cycle)."""
from aerospike_sdk.sync.transactional_session import TransactionalSession # noqa: PLC0415
return TransactionalSession
[docs]
def create_index(
self,
dataset: DataSet,
index_name: str,
bin_name: Optional[str] = None,
index_type: Optional[IndexType] = None,
collection_type: Optional[CollectionIndexType] = None,
ctx: Optional[List[CTX]] = None,
expression: Optional[Union[str, FilterExpression]] = None,
) -> IndexTask:
"""Create a secondary index in one call (synchronous).
Same arguments and rules as the async verb; a dataset and a name alone
create a set index.
Returns:
An :class:`~aerospike_native.IndexTask`; call
``wait_till_complete_blocking()`` before querying through the index.
See Also:
:meth:`aerospike_sdk.aio.session.Session.create_index`
"""
builder = self.index(dataset)._configure(
index_name, bin_name, index_type, collection_type, ctx, expression,
)
return builder.create()
[docs]
def drop_index(self, dataset: DataSet, index_name: str) -> DropIndexTask:
"""Drop a secondary index by name (synchronous).
See Also:
:meth:`aerospike_sdk.aio.session.Session.drop_index`
"""
return self.index(dataset).named(index_name).drop()
[docs]
def register_udf(
self,
body: bytes | str,
server_path: str,
language: UDFLang = UDFLang.LUA,
*,
policy: Optional["AdminPolicy"] = None,
) -> "RegisterTask":
"""Register a UDF module from bytes (synchronous).
Exposed here so the ``ClusterDefinition`` ➜ ``Cluster`` ➜ ``Session``
path is self-sufficient.
Raises:
RuntimeError: If not connected.
AerospikeError: On cluster errors (via PNC).
See Also:
:meth:`aerospike_sdk.aio.session.Session.register_udf`
"""
return self._client._register_udf(body, server_path, language, policy=policy)
[docs]
def register_udf_from_file(
self,
client_path: str,
server_path: str,
language: UDFLang = UDFLang.LUA,
*,
policy: Optional["AdminPolicy"] = None,
) -> "RegisterTask":
"""Register a UDF module from a local file (synchronous).
Raises:
RuntimeError: If not connected.
AerospikeError: On cluster errors (via PNC).
See Also:
:meth:`aerospike_sdk.aio.session.Session.register_udf_from_file`
"""
return self._client._register_udf_from_file(
client_path, server_path, language, policy=policy)
[docs]
def register_udf_from_resource(
self,
package: str,
resource: str,
server_path: str,
language: UDFLang = UDFLang.LUA,
*,
policy: Optional["AdminPolicy"] = None,
) -> "RegisterTask":
"""Register a UDF from a Python package resource (synchronous).
Raises:
RuntimeError: If not connected.
AerospikeError: On cluster errors (via PNC).
See Also:
:meth:`aerospike_sdk.aio.session.Session.register_udf_from_resource`
"""
return self._client._register_udf_from_resource(
package, resource, server_path, language, policy=policy)
[docs]
def remove_udf(
self,
server_path: str,
*,
policy: Optional["AdminPolicy"] = None,
) -> "UdfRemoveTask":
"""Remove a UDF module from the cluster (synchronous).
Raises:
RuntimeError: If not connected.
AerospikeError: On cluster errors (via PNC).
See Also:
:meth:`aerospike_sdk.aio.session.Session.remove_udf`
"""
return self._client._remove_udf(server_path, policy=policy)
[docs]
def list_udf(self) -> list[dict[str, str]]:
"""List the UDF modules registered on the cluster (synchronous).
Raises:
RuntimeError: If not connected.
AerospikeError: On cluster errors (via PNC).
See Also:
:meth:`aerospike_sdk.aio.session.Session.list_udf`
"""
return self._client._list_udf()
[docs]
def list_indexes(self) -> list[dict[str, str]]:
"""List the secondary indexes defined on the cluster (synchronous).
Raises:
RuntimeError: If not connected.
AerospikeError: On cluster errors (via PNC).
See Also:
:meth:`aerospike_sdk.aio.session.Session.list_indexes`
"""
return self._client._list_indexes()
[docs]
def do_in_transaction(
self,
operation: "typing.Callable[[TransactionalSession], typing.Any]",
*,
max_attempts: Optional[int] = None,
sleep_between_retries: Optional[Union[float, timedelta]] = None,
) -> Any:
"""Run a callable inside a retrying multi-record transaction (synchronous).
Args:
operation: Callable accepting a :class:`TransactionalSession`.
max_attempts: Maximum total attempts (initial + retries). Must be
``>= 1``. Omit to use the cluster's
:class:`~aerospike_sdk.policy.system_settings.TransactionSettings`.
sleep_between_retries: How long to wait before retrying a
blocked record or failed commit, as seconds or a
:class:`datetime.timedelta`; each wait is jittered +/- 50%.
A version mismatch always retries immediately. Omit to use
the cluster's transaction settings; pass ``0`` to never wait.
Returns:
Whatever ``operation`` returns on the successful attempt.
Raises:
ValueError: If ``max_attempts`` is given and is less than 1.
CommitError: The commit's roll-forward was abandoned (not
retried: the server will eventually commit those writes).
AerospikeError: The last-seen transient error after retries
are exhausted, or any non-transient error from ``operation``.
Example::
def transfer(tx):
tx.upsert(src).put({"balance": 90}).execute()
tx.upsert(dst).put({"balance": 110}).execute()
session.do_in_transaction(transfer)
See Also:
:meth:`aerospike_sdk.sync.transactional_session.TransactionalSession.do_in_transaction`:
Nested calls join the transaction already in progress.
"""
attempts, sleep_seconds = resolve_retry_plan(
self._client._sdk_settings.transactions,
max_attempts,
sleep_between_retries,
)
# One transaction for the usage counters however many attempts it
# takes; the per-attempt sessions do not count themselves.
usage.record_transaction(self._client)
last_exc: Optional[BaseException] = None
for attempt in range(attempts):
try:
tx = self.transaction()
tx._count_on_enter = False
with tx as tx_session:
return operation(tx_session)
except AerospikeError as exc:
last_exc = exc
if not is_retryable_txn_error(exc):
raise
if attempt + 1 >= attempts:
raise
delay = retry_delay_seconds(exc, sleep_seconds)
if delay > 0:
time.sleep(delay)
assert last_exc is not None
raise last_exc
# -- Write-verb factories -------------------------------------------------
# The eight public write verbs (upsert/insert/.../exists) and the
# single-key predicate live on `SessionBase`; these two helpers stay
# per-leaf because they construct the sync builders and wire the sync
# namespace-mode resolver.
def _fast_write_segment(self, op_type: str, key: Key) -> WriteSegmentBuilder:
"""Single-key fast-path write segment (sync)."""
return _SingleKeyWriteSegment(
client=self._pnc_client,
key=key,
op_type=op_type,
behavior=self._behavior,
write_policy=self._cached_write_policy,
read_policy=self._cached_read_policy,
write_policy_sc=self._cached_write_policy_sc,
read_policy_sc=self._cached_read_policy_sc,
txn=self._txn,
namespace_mode_resolver=None,
namespace_mode_resolver_blocking=self._resolve_namespace_mode_blocking,
sdk_client=self._client,
)
def _dataset_write_builder(self, op_type: str, dataset: DataSet) -> DataSetWriteBuilder:
"""Dataset-scoped tabular write; rows arrive through ``bins(...)``."""
qb = self._build_sync_query_builder(dataset=dataset, key=None, keys=None)
self._bind_txn(qb)
return DataSetWriteBuilder(qb, op_type, dataset)
def _build_write_segment(
self, op_type: str, arg1: Union[Key, List[Key]], *more_keys: Key,
) -> WriteSegmentBuilder:
"""Build a multi-key write segment via the sync QueryBuilder."""
if isinstance(arg1, Key):
if more_keys:
qb = self._build_sync_query_builder(
dataset=None, key=None, keys=[arg1, *more_keys],
)
else:
qb = self._build_sync_query_builder(dataset=None, key=arg1, keys=None)
elif isinstance(arg1, list) and not more_keys:
if not arg1:
raise ValueError("keys list cannot be empty")
qb = self._build_sync_query_builder(dataset=None, key=None, keys=list(arg1))
else:
raise TypeError(f"Expected Key or List[Key], got {type(arg1)}")
self._bind_txn(qb)
qb._op_type = op_type
return WriteSegmentBuilder(qb)