Connecting to a Cluster¶

Basic Connection¶

ClusterDefinition is the entry point for every connection. Connect it to obtain a Cluster, then create sessions from that cluster:

Async

from aerospike_sdk import ClusterDefinition

async with ClusterDefinition("localhost", 3000).connect() as cluster:
    session = cluster.create_session()
    # ... use session ...

Sync

from aerospike_sdk.sync import ClusterDefinition

with ClusterDefinition("localhost", 3000).connect() as cluster:
    session = cluster.create_session()
    # ... use session ...

The Cluster returned by connect() supports the context-manager protocol, which automatically closes the connection on exit.

Sync usage¶

The same surface is available without asyncio — no async/await, no event loop — useful for sync codebases or when a dependency forbids asyncio. Connect through the sync ClusterDefinition and every builder chain reads the same, minus the await:

from aerospike_sdk import Behavior, DataSet
from aerospike_sdk.sync import ClusterDefinition


def main():
    with ClusterDefinition("localhost", 3000).connect() as cluster:
        session = cluster.create_session(Behavior.DEFAULT)
        users = DataSet.of("test", "users")

        # High-level key-value writes
        session.upsert(users.id(1)).put({"name": "Alice", "age": 28, "country": "UK"}).execute()
        session.upsert(users.id(2)).put({"name": "Bob", "age": 35, "country": "US"}).execute()

        # Filtered query with AEL — same builder API as async
        results = (
            session.query(users)
            .where("$.age > 25 and $.country == 'US'")
            .execute()
        )
        for row in results:
            if row.is_ok and row.record is not None:
                print(row.record.bins)


main()

The sync surface is an independent synchronous implementation that calls PNC’s blocking entry points directly — there is no per-call event loop and no per-thread loop runner. Sessions, behaviors, builders, and AEL filters are identical to the async surface.

Advanced Configuration¶

ClusterDefinition exposes a fluent builder for credentials, alternate-access, IP mapping, TLS, rack awareness, and more. Credentials come in four forms: with_native_credentials (internal users), with_external_credentials (LDAP and similar, password sent under TLS), with_external_insecure_credentials (the same without TLS, for trusted networks only), and with_certificate_credentials (PKI). Client-wide connection limits and timeouts, including tend_timeout and login_timeout, are SystemSettings passed through with_system_settings:

from aerospike_sdk import ClusterDefinition

cluster_def = (
    ClusterDefinition("localhost", 3000)
    .with_native_credentials("username", "password")
    .using_services_alternate()
    .ip_map({"10.0.0.1": "3.72.54.187"})
)

async with cluster_def.connect() as cluster:
    session = cluster.create_session()
    # ...

Alternate access addresses¶

using_services_alternate() makes peer discovery follow each node’s alternate-access-address rather than its standard service address. Enable it only against a cluster that publishes those addresses: against one that does not, the peer list comes back empty, the client falls back to a single node, and every key outside that node’s partitions fails to route.

The setting defaults to on whenever AEROSPIKE_USE_SERVICES_ALTERNATE is truthy in the environment, so pass False explicitly to force it off regardless of the environment:

cluster_def = ClusterDefinition("localhost", 3000).using_services_alternate(False)

TLS¶

Server-side TLS with CA certificate verification:

cluster_def = (
    ClusterDefinition("localhost", 4333)
    .with_tls_config(tls_name="myTlsName", ca_file="/path/to/ca.pem")
    .with_native_credentials("username", "password")
    .using_services_alternate()
)

async with cluster_def.connect() as cluster:
    session = cluster.create_session()
    # ... use session ...

Mutual TLS (mTLS) with client certificate authentication:

cluster_def = (
    ClusterDefinition("localhost", 4333)
    .with_tls_config(
        tls_name="myTlsName",
        ca_file="/path/to/ca.pem",
        client_cert_file="/path/to/client-cert.pem",
        client_key_file="/path/to/client-key.pem",
    )
    .with_native_credentials("username", "password")
    .using_services_alternate()
)

async with cluster_def.connect() as cluster:
    session = cluster.create_session()
    # ... use session ...

Note

The tls_name must match the server’s configured TLS name for certificate validation. It is applied to every host that has no tls_name of its own. Each with_tls_config() call replaces any earlier TLS settings.

To restrict the handshake, pass protocols=["TLSv1.3"] or a list of cipher-suite names in ciphers=. An empty list raises ValueError right away; omit the argument to allow the defaults. At connect, an unknown name or a combination that leaves no usable suite (TLS 1.2 with only TLS 1.3 suites, for example) raises ValueError.

For certificate (PKI) authentication, pass client_cert_file and client_key_file here and call with_certificate_credentials(); connect refuses PKI without a client certificate.

TLS for the login exchange only, with cleartext data connections, for trusted networks where data-plane encryption is not worth its cost:

cluster_def = (
    ClusterDefinition("localhost", 4333)
    .with_tls_config(
        tls_name="myTlsName", ca_file="/path/to/ca.pem", for_login_only=True,
    )
    .with_native_credentials("username", "password")
    .using_services_alternate()
)

The credential exchange runs over TLS; that connection is then closed, and every later connection is opened in cleartext at the node’s cleartext service address. No socket is ever downgraded. The node must advertise a cleartext service port, and credentials are required: login-only without authentication would only mean “no TLS at all”, so the client refuses it.

Rack Awareness¶

cluster = await (
    ClusterDefinition("localhost", 3000)
    .preferring_racks(1, 2)
    .connect()
)

Application Identifier¶

Tag this client’s traffic with an application identifier via app_id. The server records it (as part of the client’s user-agent), letting operators attribute load per calling application. It is separate from the client-library identifier the SDK reports automatically.

cluster = await (
    ClusterDefinition("localhost", 3000)
    .app_id("billing-service")
    .connect()
)

Metrics snapshots carry the same identifier, so client-side latency and errors join to the server’s view of the same application. Leave it unset and the user-agent reports not-set, while the snapshot reports the authenticated user if there is one, falling back to not-set as well — the snapshot’s field is never empty. The authenticated-user fallback applies to metrics only; it is not repeated in the user-agent, where the server already knows the connection’s user. See Client Metrics.

SDK Configuration File¶

Connection pool sizing, circuit-breaker thresholds, tend interval, and named operation-policy behaviors can also come from a YAML file (the AEROSPIKE_SDK_CONFIG_URL environment variable), read at connect() and hot-reloaded on change. See Dynamic SDK Configuration.

Checking Server Capabilities¶

Every feature works on every supported server (8.2.0 and later), but a cluster in the middle of a rolling upgrade can briefly include nodes on an older build. You can guard a feature before using it there, rather than letting a call fail at runtime. The Cluster reports a feature as supported only when every connected node supports it, so you always guard against the cluster’s least-capable node.

async with ClusterDefinition("localhost", 3000).connect() as cluster:
    if cluster.supports_string_operations():
        await session.upsert(key).bin("s").str_append("!").execute()
    else:
        ...  # fall back

    version = cluster.server_version()   # minimum Version across the cluster
    if version is not None and (version.major, version.minor, version.patch) >= (8, 2, 0):
        ...

The probes are supports_ael(), supports_query_operations(), supports_string_operations(), supports_query_selection(), and server_version() (the minimum version across all nodes, or None if the cluster reports no nodes). They read the client’s own view of the cluster, so they are plain methods on both the async and sync Cluster, and the answer reflects current cluster membership.

Inspecting Cluster Nodes¶

server_version() gives the cluster minimum. To see each node, list them with nodes(), or look one up by name with get_node(). Each Node reports its name, address, and version, and can run an info command against that node alone:

async with ClusterDefinition("localhost", 3000).connect() as cluster:
    print(f"{cluster.cluster_name}: {cluster.node_names()}")

    versions = {node.name: str(node.version) for node in cluster.nodes()}
    if len(set(versions.values())) > 1:
        print(f"rolling upgrade in progress: {versions}")

    node = cluster.get_node("BB9D4EB574A8DA6")
    print(await node.info("statistics"))

cluster.cluster_name is the name the servers report, or None when they have none configured. If you connected with ClusterDefinition.cluster_name(...), it is the same name, since nodes reporting any other name are rejected. Unknown names passed to get_node() raise InvalidNodeError.

These reads come from the client’s own view of the cluster, so they are plain methods on both the async and sync Cluster. Only Node.info() goes to the server, and it is the one you await on the async side.

For an info command on every node at once, with parsed results, use InfoCommands and its *_per_node variants.

Sessions¶

A Session binds a connected cluster to a set of policy defaults via a Behavior. All reads and writes go through a session.

from aerospike_sdk import Behavior

session = cluster.create_session(Behavior.DEFAULT)
fast_session = cluster.create_session(Behavior.READ_FAST)
consistent_session = cluster.create_session(Behavior.STRICTLY_CONSISTENT)

# or derive from a session you already hold; the original is unchanged
fast_session = session.session_for(Behavior.READ_FAST)

Tip

Create different sessions for different workloads. A “fast read” session with short timeouts and a “batch import” session with longer timeouts can coexist on the same cluster.

Behaviors¶

Predefined behaviors:

Behavior	Description
`Behavior.DEFAULT`	Balanced defaults
`Behavior.READ_FAST`	Low-latency reads
`Behavior.STRICTLY_CONSISTENT`	Strong consistency mode
`Behavior.FAST_RACK_AWARE`	Prefer local rack for reads

Custom behaviors via derivation:

from datetime import timedelta

from aerospike_sdk.policy import Settings

my_behavior = Behavior.DEFAULT.derive_with_changes(
    "generous-timeouts",
    all=Settings(total_timeout=timedelta(seconds=5), max_retries=3),
)

Each change names the operations it applies to: all= covers every operation, and narrower scopes such as reads=, reads_batch=, or writes_sc= override it for their operations.

DataSets¶

A DataSet represents a namespace + set pair and is a convenient key factory:

from aerospike_sdk import DataSet

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

key = users.id(42)                       # single key
keys = users.ids(1, 2, 3)               # list of keys
digest_key = users.id_from_digest(b"...")  # key from raw digest

Pass datasets to session methods to avoid repeating namespace/set strings:

stream = await session.query(users).execute()
await session.upsert(users.id(1)).put({"name": "Alice"}).execute()