From 293bcd87725476d0c78c9a900d53e9ba4feea09d Mon Sep 17 00:00:00 2001 From: Paul Brackin Date: Tue, 27 Oct 2026 12:37:35 -0700 Subject: [PATCH] CLIENT-5776 Accept a list of Host as client seeds new_client, new_client_blocking and _LocalClient take a seed string or a sequence of the new Host class; a host list goes to the core as-is, so an IPv6 address or TLS name needs no bracketing or quoting. --- python/aerospike_native/_native.pyi | 61 +++++++- python/tests/integration/blocking_test.py | 21 +++ python/tests/integration/client_test.py | 14 +- python/tests/unit/host_test.py | 77 +++++++++ src/blocking.rs | 13 +- src/host.rs | 182 ++++++++++++++++++++++ src/lib.rs | 19 ++- 7 files changed, 369 insertions(+), 18 deletions(-) create mode 100644 python/tests/unit/host_test.py create mode 100644 src/host.rs diff --git a/python/aerospike_native/_native.pyi b/python/aerospike_native/_native.pyi index a7306b2..48b8ae0 100644 --- a/python/aerospike_native/_native.pyi +++ b/python/aerospike_native/_native.pyi @@ -58,6 +58,7 @@ __all__ = [ "HLLWriteFlags", "Histogram", "HllOperation", + "Host", "IndexTask", "IndexType", "Key", @@ -4251,6 +4252,60 @@ class HllOperation: (``HLLWriteFlags.X | HLLWriteFlags.Y``), or a raw ``int``. """ +@typing.final +class Host: + r""" + Seed address of a cluster node. + + Pass a list of these to :func:`new_client` (or :func:`new_client_blocking`) + instead of a seed string. Each field is taken as-is, so an IPv6 address + or a TLS name needs no quoting or bracketing. + + Example:: + + seeds = [Host("10.0.0.1", 3000), Host("fd00::2", 3000)] + client = await new_client(ClientPolicy(), seeds) + """ + @property + def name(self) -> builtins.str: + r""" + Hostname or IP address. + """ + @property + def port(self) -> builtins.int: + r""" + Service port. + """ + @property + def tls_name(self) -> typing.Optional[builtins.str]: + r""" + TLS certificate name, or ``None`` for a host reached without TLS. + """ + def __eq__(self, other: builtins.object, /) -> builtins.bool: ... + def __hash__(self) -> builtins.int: ... + def __new__(cls, name: builtins.str, port: builtins.int, tls_name: typing.Optional[builtins.str] = None) -> _native.Host: + r""" + Create a Host. + + Args: + name: Hostname or IP address. An IPv6 address is given bare, + e.g. ``"fd00::2"``, not ``"[fd00::2]"``. + port: Service port. + tls_name: Name the node's TLS certificate must carry. ``None`` + (or blank) for a host reached without TLS. + + Returns: + Host + + Raises: + ValueError: If *name* is empty or bracketed. + """ + def __str__(self) -> builtins.str: + r""" + Seed-string form: ``name[:tls_name]:port``, IPv6 bracketed. + """ + def __repr__(self) -> builtins.str: ... + class IndexTask: def query_status(self) -> typing.Awaitable[TaskStatus]: ... def wait_till_complete(self, sleep_time: builtins.float = 0.25, timeout: typing.Optional[builtins.float] = 60.0) -> typing.Awaitable[bool]: ... @@ -7057,7 +7112,7 @@ class _LocalClient: Server-reported cluster name (see :attr:`Client.server_cluster_name`). """ - def __new__(cls, policy: _native.ClientPolicy, seeds: builtins.str) -> _native._LocalClient: ... + def __new__(cls, policy: _native.ClientPolicy, seeds: builtins.str | typing.Sequence[_native.Host]) -> _native._LocalClient: ... def get_blocking(self, key: _native.Key, bins: typing.Optional[typing.Sequence[builtins.str]] = None, *, policy: typing.Optional[_native.ReadPolicy] = None, policy_sc: typing.Optional[_native.ReadPolicy] = None, filter_expression: typing.Optional[_native.FilterExpression] = None, txn: typing.Optional[_native.Txn] = None) -> _native.Record: ... def put_blocking(self, key: _native.Key, bins: typing.Mapping[builtins.str, typing.Any], *, policy: typing.Optional[_native.WritePolicy] = None, policy_sc: typing.Optional[_native.WritePolicy] = None, txn: typing.Optional[_native.Txn] = None) -> None: ... def delete_blocking(self, key: _native.Key, *, policy: typing.Optional[_native.WritePolicy] = None) -> builtins.bool: ... @@ -7814,9 +7869,9 @@ def has_any_write_op(operations: typing.Sequence[typing.Any]) -> builtins.bool: or the server rejects the per-node group. """ -def new_client(policy: _native.ClientPolicy, seeds: builtins.str) -> typing.Awaitable[Client]: ... +def new_client(policy: _native.ClientPolicy, seeds: builtins.str | typing.Sequence[_native.Host]) -> typing.Awaitable[Client]: ... -def new_client_blocking(policy: _native.ClientPolicy, seeds: builtins.str) -> _native.Client: +def new_client_blocking(policy: _native.ClientPolicy, seeds: builtins.str | typing.Sequence[_native.Host]) -> _native.Client: r""" Synchronously create and connect a Client. diff --git a/python/tests/integration/blocking_test.py b/python/tests/integration/blocking_test.py index b4fbe0c..ab79403 100644 --- a/python/tests/integration/blocking_test.py +++ b/python/tests/integration/blocking_test.py @@ -50,6 +50,7 @@ CTX, Filter, FilterExpression as fe, + Host, IndexType, Key, LoopVarPart, @@ -770,6 +771,26 @@ def test_blocking_core_panic_is_catchable(connect, aerospike_host, use_services_ client.close_blocking() +@pytest.mark.parametrize( + "client_type", [new_client_blocking, _LocalClient], ids=["client", "local_client"], +) +def test_blocking_connect_with_hosts(client_type, aerospike_host, use_services_alternate): + """Each blocking constructor connects from a list of Host.""" + name, port = aerospike_host.rsplit(":", 1) + cp = ClientPolicy() + cp.use_services_alternate = use_services_alternate + client = client_type(cp, [Host(name, int(port))]) + key = Key("test", "blocking", "hosts-1") + try: + client.put_blocking(key, {"v": 1}, policy=WritePolicy()) + assert client.get_blocking(key, policy=ReadPolicy()).bins["v"] == 1 + finally: + client.delete_blocking(key, policy=WritePolicy()) + # The local client owns its runtime and shuts down on drop. + if not isinstance(client, _LocalClient): + client.close_blocking() + + def test_blocking_create_set_index(aerospike_host, use_services_alternate): cp = ClientPolicy() cp.use_services_alternate = use_services_alternate diff --git a/python/tests/integration/client_test.py b/python/tests/integration/client_test.py index 808d013..9eb30af 100644 --- a/python/tests/integration/client_test.py +++ b/python/tests/integration/client_test.py @@ -17,7 +17,7 @@ import pytest -from aerospike_native import new_client, ClientPolicy +from aerospike_native import new_client, ClientPolicy, Host from aerospike_native.exceptions import ConnectionError, ResultCode async def test_connect(): @@ -28,6 +28,18 @@ async def test_connect(): client = await new_client(cp, os.environ.get("AEROSPIKE_HOST", "localhost:3000")) assert client is not None +async def test_connect_with_hosts(): + """A list of Host connects, and seeds() reports it in seed-string form.""" + name, port = os.environ.get("AEROSPIKE_HOST", "localhost:3000").rsplit(":", 1) + cp = ClientPolicy() + cp.use_services_alternate = True + client = await new_client(cp, [Host(name, int(port))]) + try: + assert client.is_connected() is True + assert client.seeds() == f"{name}:{port}" + finally: + await client.close() + async def test_failed_connect(): """Test basic client connection failure.""" client = None diff --git a/python/tests/unit/host_test.py b/python/tests/unit/host_test.py new file mode 100644 index 0000000..33d9840 --- /dev/null +++ b/python/tests/unit/host_test.py @@ -0,0 +1,77 @@ +# Copyright 2023-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. + +"""Unit tests for Host and the seed argument of the client constructors.""" + +import pytest +from aerospike_native import ClientPolicy, Host, new_client_blocking + + +class TestHost: + """Host construction, formatting, and value semantics.""" + + def test_fields(self): + host = Host("db1.example.com", 4333, tls_name="db1") + assert (host.name, host.port, host.tls_name) == ("db1.example.com", 4333, "db1") + + def test_blank_tls_name_is_none(self): + assert Host("db1", 3000, tls_name=" ").tls_name is None + + @pytest.mark.parametrize( + ("host", "seed"), + [ + (Host("10.0.0.1", 3000), "10.0.0.1:3000"), + (Host("db1", 4333, tls_name="tls1"), "db1:tls1:4333"), + (Host("fd00::2", 3000), "[fd00::2]:3000"), + (Host("fd00::2", 4333, tls_name="tls1"), "[fd00::2]:tls1:4333"), + ], + ) + def test_str_is_seed_entry(self, host, seed): + assert str(host) == seed + + def test_repr(self): + assert repr(Host("fd00::2", 3000)) == "Host('fd00::2', 3000)" + assert repr(Host("db1", 4333, tls_name="tls1")) == "Host('db1', 4333, tls_name='tls1')" + + def test_equality_and_hash(self): + assert Host("db1", 3000) == Host("db1", 3000) + assert Host("db1", 3000) != Host("db1", 3000, tls_name="tls1") + assert len({Host("db1", 3000), Host("db1", 3000)}) == 1 + + def test_immutable(self): + with pytest.raises(AttributeError): + Host("db1", 3000).port = 3001 # type: ignore[misc] + + @pytest.mark.parametrize("name", ["", "[fd00::2]"]) + def test_rejects_invalid_name(self, name): + with pytest.raises(ValueError): + Host(name, 3000) + + def test_rejects_out_of_range_port(self): + with pytest.raises(OverflowError): + Host("db1", 70000) + + +class TestSeedsArgument: + """Seed values rejected before any connection attempt.""" + + def test_empty_host_list(self): + with pytest.raises(ValueError, match="at least one Host"): + new_client_blocking(ClientPolicy(), []) + + @pytest.mark.parametrize("seeds", [["db1:3000"], 3000, None]) + def test_wrong_type(self, seeds): + with pytest.raises(TypeError, match="str or a sequence of Host"): + new_client_blocking(ClientPolicy(), seeds) diff --git a/src/blocking.rs b/src/blocking.rs index 661e3ab..9f81e2b 100644 --- a/src/blocking.rs +++ b/src/blocking.rs @@ -37,6 +37,7 @@ use pyo3_stub_gen::derive::gen_stub_pyfunction; use tokio::runtime::Runtime; use crate::errors::{panic_to_pyerr, RustClientError}; +use crate::host::Seeds; use crate::policies::ClientPolicy; use crate::Client; @@ -121,21 +122,21 @@ where pub(crate) fn new_client_blocking( py: Python<'_>, policy: ClientPolicy, - seeds: String, + seeds: Seeds, ) -> PyResult { let as_policy = policy._as.clone(); - let as_seeds = seeds.clone(); + let seed_string = seeds.to_seed_string(); let cluster_name = as_policy.cluster_name.clone(); + log::debug!(target: "aerospike_native", "connecting (blocking) to {}", seed_string); let raw = run_blocking(py, async move { - log::debug!(target: "aerospike_native", "connecting (blocking) to {}", as_seeds); - aerospike_core::Client::new(&as_policy, &as_seeds) + aerospike_core::Client::new(&as_policy, seeds.as_to_hosts()) .await .map_err(|e| PyErr::from(RustClientError(e))) })?; - log::debug!(target: "aerospike_native", "connected (blocking) to {}", seeds); + log::debug!(target: "aerospike_native", "connected (blocking) to {}", seed_string); Ok(Client { _as: Arc::new(raw), - seeds, + seeds: seed_string, cluster_name, bridge: None, }) diff --git a/src/host.rs b/src/host.rs new file mode 100644 index 0000000..5409362 --- /dev/null +++ b/src/host.rs @@ -0,0 +1,182 @@ +// Copyright 2023-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. + +use aerospike_core::ToHosts; +use pyo3::exceptions::{PyTypeError, PyValueError}; +use pyo3::prelude::*; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::{PyStubType, TypeInfo}; + + //////////////////////////////////////////////////////////////////////////////////////////// + // + // Host + // + //////////////////////////////////////////////////////////////////////////////////////////// + + /// Seed address of a cluster node. + /// + /// Pass a list of these to :func:`new_client` (or :func:`new_client_blocking`) + /// instead of a seed string. Each field is taken as-is, so an IPv6 address + /// or a TLS name needs no quoting or bracketing. + /// + /// Example:: + /// + /// seeds = [Host("10.0.0.1", 3000), Host("fd00::2", 3000)] + /// client = await new_client(ClientPolicy(), seeds) + #[gen_stub_pyclass(module = "_native")] + #[pyclass(from_py_object, module = "_native", frozen, eq, hash)] + #[derive(Clone, PartialEq, Eq, Hash)] + pub struct Host { + pub(crate) _as: aerospike_core::Host, + } + + #[gen_stub_pymethods] + #[pymethods] + impl Host { + /// Create a Host. + /// + /// Args: + /// name: Hostname or IP address. An IPv6 address is given bare, + /// e.g. ``"fd00::2"``, not ``"[fd00::2]"``. + /// port: Service port. + /// tls_name: Name the node's TLS certificate must carry. ``None`` + /// (or blank) for a host reached without TLS. + /// + /// Returns: + /// Host + /// + /// Raises: + /// ValueError: If *name* is empty or bracketed. + #[new] + #[pyo3(signature = (name, port, tls_name=None))] + pub fn new(name: &str, port: u16, tls_name: Option<&str>) -> PyResult { + if name.is_empty() { + return Err(PyValueError::new_err("Host name must not be empty")); + } + // A bracketed name would be resolved literally, brackets included, + // and fail as an unknown hostname far from the cause. + if name.starts_with('[') { + return Err(PyValueError::new_err(format!( + "Host name {name:?} is bracketed; give the IPv6 address bare" + ))); + } + let host = match tls_name { + Some(tls_name) => aerospike_core::Host::new_tls(name, tls_name, port), + None => aerospike_core::Host::new(name, port), + }; + Ok(Host { _as: host }) + } + + /// Hostname or IP address. + #[getter] + pub fn name(&self) -> &str { + &self._as.name + } + + /// Service port. + #[getter] + pub fn port(&self) -> u16 { + self._as.port + } + + /// TLS certificate name, or ``None`` for a host reached without TLS. + #[getter] + pub fn tls_name(&self) -> Option<&str> { + self._as.tls_name.as_deref() + } + + /// Seed-string form: ``name[:tls_name]:port``, IPv6 bracketed. + pub fn __str__(&self) -> String { + seed_entry(&self._as) + } + + pub fn __repr__(&self) -> String { + match &self._as.tls_name { + Some(tls_name) => format!( + "Host('{}', {}, tls_name='{}')", + self._as.name, self._as.port, tls_name + ), + None => format!("Host('{}', {})", self._as.name, self._as.port), + } + } + } + + /// Format *host* as one seed-string entry, bracketing an IPv6 name so its + /// colons are not read as field separators. + fn seed_entry(host: &aerospike_core::Host) -> String { + let (open, close) = if host.name.contains(':') { ("[", "]") } else { ("", "") }; + match &host.tls_name { + Some(tls_name) => format!("{open}{}{close}:{tls_name}:{}", host.name, host.port), + None => format!("{open}{}{close}:{}", host.name, host.port), + } + } + + //////////////////////////////////////////////////////////////////////////////////////////// + // + // Seeds + // + //////////////////////////////////////////////////////////////////////////////////////////// + + /// The seed argument of the client constructors: a seed string, parsed by + /// the core, or a non-empty sequence of :class:`Host`, passed through. + pub enum Seeds { + Text(String), + Hosts(Vec), + } + + impl Seeds { + pub(crate) fn as_to_hosts(&self) -> &(dyn ToHosts + Send + Sync) { + match self { + Seeds::Text(text) => text, + Seeds::Hosts(hosts) => hosts, + } + } + + /// Seed-string form, as reported by :meth:`Client.seeds`. + pub(crate) fn to_seed_string(&self) -> String { + match self { + Seeds::Text(text) => text.clone(), + Seeds::Hosts(hosts) => { + hosts.iter().map(seed_entry).collect::>().join(",") + } + } + } + } + + impl<'a, 'py> FromPyObject<'a, 'py> for Seeds { + type Error = PyErr; + + fn extract(obj: Borrowed<'a, 'py, PyAny>) -> Result { + if let Ok(text) = obj.extract::() { + return Ok(Seeds::Text(text)); + } + let hosts: Vec = obj.extract().map_err(|_| { + PyTypeError::new_err("seeds must be a str or a sequence of Host") + })?; + // With no seeds the core raises a connection error that does not + // say why, or, with `fail_if_not_connected` off, returns a client + // that can never reach a node. + if hosts.is_empty() { + return Err(PyValueError::new_err("seeds must contain at least one Host")); + } + Ok(Seeds::Hosts(hosts.into_iter().map(|host| host._as).collect())) + } + } + + impl PyStubType for Seeds { + fn type_output() -> TypeInfo { + String::type_input() | Vec::::type_input() + } + } diff --git a/src/lib.rs b/src/lib.rs index fb2f590..0b5f2b0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -53,6 +53,7 @@ mod runtime; mod client_runtime; mod tasks; mod tls; +mod host; mod record; mod cdt; mod expressions; @@ -78,6 +79,7 @@ pub use query_plan::*; pub use cluster::*; pub use server_error::*; pub use tls::*; +pub use host::*; pub use string_ops::*; pub use metrics::*; @@ -98,9 +100,9 @@ use crate::operations::{ #[gen_stub_pyfunction(module = "_native")] #[pyfunction] #[gen_stub(override_return_type(type_repr="typing.Awaitable[Client]", imports=("typing")))] - pub fn new_client(py: Python, policy: ClientPolicy, seeds: String) -> PyResult> { + pub fn new_client(py: Python, policy: ClientPolicy, seeds: Seeds) -> PyResult> { let as_policy = policy._as.clone(); - let as_seeds = seeds.clone(); + let seed_string = seeds.to_seed_string(); let cluster_name = as_policy.cluster_name.clone(); // Capture the loop the caller is awaiting on; this becomes the bridge's // owning loop and every subsequent op on this Client must run on it. @@ -132,15 +134,15 @@ use crate::operations::{ // queuing I/O serially behind the global reactor (~5ms latency). let bridge_for_construction = bridge.clone(); Ok(completion::batched_future_into_py(&bridge_for_construction, py, async move { - log::debug!(target: "aerospike_native", "connecting to {}", as_seeds); - let c = aerospike_core::Client::new(&as_policy, &as_seeds) + log::debug!(target: "aerospike_native", "connecting to {}", seed_string); + let c = aerospike_core::Client::new(&as_policy, seeds.as_to_hosts()) .await .map_err(|e| PyErr::from(RustClientError(e)))?; - log::debug!(target: "aerospike_native", "connected to {}", seeds); + log::debug!(target: "aerospike_native", "connected to {}", seed_string); let res = Client { _as: Arc::new(c), - seeds: seeds.clone(), + seeds: seed_string, cluster_name, bridge: Some(bridge), }; @@ -433,7 +435,7 @@ use crate::operations::{ #[pymethods] impl LocalClient { #[new] - pub fn new(policy: ClientPolicy, seeds: String) -> PyResult { + pub fn new(policy: ClientPolicy, seeds: Seeds) -> PyResult { let rt = tokio::runtime::Builder::new_current_thread() .enable_io() .enable_time() @@ -444,7 +446,7 @@ use crate::operations::{ let as_policy = policy._as.clone(); let cluster_name = as_policy.cluster_name.clone(); let client = crate::blocking::block_on(&rt, async move { - aerospike_core::Client::new(&as_policy, &seeds).await + aerospike_core::Client::new(&as_policy, seeds.as_to_hosts()).await })?.map_err(|e| PyErr::from(RustClientError(e)))?; Ok(LocalClient { rt, @@ -5302,6 +5304,7 @@ fn _native(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; #[cfg(feature = "tls")] m.add_class::()?; + m.add_class::()?; m.add_function(wrap_pyfunction!(new_client, m)?)?; m.add_function(wrap_pyfunction!(crate::blocking::new_client_blocking, m)?)?;