From c8157a2b54693fe78e29150caeb8c069a949ac9d Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:14:49 +0800 Subject: [PATCH 1/7] feat(python): expose TCP client configuration to Python IggyClient accepted only a server address, so auto-login and reconnection tuning were unreachable from Python. Without credentials to replay, the SDK's own session recovery never fires and a dropped session surfaces as Unauthenticated on the next call, leaving the application to hand-roll a connect/login/probe loop. TcpConfig mirrors TcpClientConfig field for field and is accepted by the IggyClient constructor alongside the existing address string. AutoLogin carries the credentials without exposing them back to Python, and TcpReconnectionConfig carries the retry policy. Credentials is re-exported from the SDK prelude because AutoLogin::Enabled cannot be constructed without naming it. Closes #3742 --- core/sdk/src/prelude.rs | 6 +- foreign/python/Cargo.toml | 1 + foreign/python/apache_iggy.pyi | 145 ++++++++++++- foreign/python/src/client.rs | 36 +++- foreign/python/src/config.rs | 368 +++++++++++++++++++++++++++++++++ foreign/python/src/consumer.rs | 17 +- foreign/python/src/duration.rs | 43 ++++ foreign/python/src/lib.rs | 6 + 8 files changed, 599 insertions(+), 23 deletions(-) create mode 100644 foreign/python/src/config.rs create mode 100644 foreign/python/src/duration.rs diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index 34f8d14844..add8cad1fd 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -51,9 +51,9 @@ pub use iggy_common::{ Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, CacheMetrics, CacheMetricsKey, ClientError, ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember, - ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, HeaderValue, - HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo, - IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage, + ConsumerKind, Credentials, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, + HeaderValue, HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, + IdentityInfo, IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage, IggyMessageHeader, IggyMessageHeaderView, IggyMessageView, IggyMessageViewIterator, IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind, PollingStrategy, diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml index 612cec38a2..aa2ee442e9 100644 --- a/foreign/python/Cargo.toml +++ b/foreign/python/Cargo.toml @@ -44,4 +44,5 @@ pyo3-async-runtimes = { version = "0.29.0", features = [ "tokio-runtime", ] } pyo3-stub-gen = "0.23.0" +secrecy = "0.10" tokio = "1.53.1" diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 5b980c2e0d..3befd0deb7 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -29,6 +29,7 @@ __all__ = [ "AutoCommit", "AutoCommitAfter", "AutoCommitWhen", + "AutoLogin", "ConsumerGroup", "ConsumerGroupDetails", "ConsumerGroupMember", @@ -38,6 +39,8 @@ __all__ = [ "ReceiveMessage", "SendMessage", "StreamDetails", + "TcpConfig", + "TcpReconnectionConfig", "Topic", "TopicDetails", "UserInfo", @@ -238,6 +241,41 @@ class AutoCommitWhen: ... +@typing.final +class AutoLogin: + r""" + The credentials replayed by the client every time it (re)connects. + + `IggyClient` only recovers a lost session when it has credentials to replay, + so a long-running consumer should pass one of the enabled variants. + """ + @property + def enabled(self) -> builtins.bool: + r""" + Whether automatic login is enabled. + """ + @property + def username(self) -> builtins.str | None: + r""" + The username to log in with, or `None` for the disabled and token variants. + """ + @staticmethod + def disabled() -> AutoLogin: + r""" + No automatic login. `login_user()` must be called by hand after every connect. + """ + @staticmethod + def username_password(username: builtins.str, password: builtins.str) -> AutoLogin: + r""" + Log in with the given username and password on every connect. + """ + @staticmethod + def personal_access_token(token: builtins.str) -> AutoLogin: + r""" + Log in with the given personal access token on every connect. + """ + def __repr__(self) -> builtins.str: ... + @typing.final class ConsumerGroup: @property @@ -314,11 +352,20 @@ class IggyClient: It wraps the RustIggyClient and provides asynchronous functionality through the contained runtime. """ - def __new__(cls, conn: builtins.str | None = None) -> IggyClient: + def __new__(cls, conn: TcpConfig | builtins.str | None = None) -> IggyClient: r""" - Constructs a new IggyClient from a TCP server address. + Constructs a new IggyClient from a TCP server address or a `TcpConfig`. This initializes a new runtime for asynchronous operations. Future versions might utilize asyncio for more Pythonic async. + + Args: + conn: Either a `host:port` address, or a `TcpConfig` carrying the full + transport configuration. Defaults to `127.0.0.1:8090` with auto-login + disabled. + + Raises: + PyRuntimeError: If the address is not a valid `host:port` pair, or if the + client cannot be built. """ @classmethod def from_connection_string(cls, connection_string: builtins.str) -> IggyClient: @@ -916,6 +963,100 @@ class StreamDetails: @property def topics_count(self) -> builtins.int: ... +@typing.final +class TcpConfig: + r""" + Configuration for the TCP transport, accepted by `IggyClient(...)`. + + Mirrors `TcpClientConfig` in the Rust SDK. Every field is keyword-only and + falls back to the same default the Rust SDK uses. + """ + @property + def server_address(self) -> builtins.str: ... + @property + def auto_login(self) -> AutoLogin: ... + @property + def reconnection(self) -> TcpReconnectionConfig: ... + @property + def heartbeat_interval(self) -> datetime.timedelta: ... + @property + def tls_enabled(self) -> builtins.bool: ... + @property + def tls_domain(self) -> builtins.str: ... + @property + def tls_ca_file(self) -> builtins.str | None: ... + @property + def tls_validate_certificate(self) -> builtins.bool: ... + @property + def nodelay(self) -> builtins.bool: ... + def __new__( + cls, + *, + server_address: builtins.str | None = None, + auto_login: AutoLogin | None = None, + reconnection: TcpReconnectionConfig | None = None, + heartbeat_interval: datetime.timedelta | None = None, + tls_enabled: builtins.bool = False, + tls_domain: builtins.str | None = None, + tls_ca_file: builtins.str | None = None, + tls_validate_certificate: builtins.bool = True, + nodelay: builtins.bool = False, + ) -> TcpConfig: + r""" + Constructs a TCP configuration, defaulting every unset field to the value + the Rust SDK uses. + + Args: + server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`. + auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. + heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + tls_enabled: Whether to connect over TLS. + tls_domain: Domain to validate the certificate against. Empty means it is + taken from `server_address`. + tls_ca_file: Path to the CA file for TLS. + tls_validate_certificate: Whether to validate the server certificate. + nodelay: Disable the Nagle algorithm for the TCP socket. + + Raises: + PyValueError: If `server_address` is not a valid `host:port` pair. + """ + def __repr__(self) -> builtins.str: ... + +@typing.final +class TcpReconnectionConfig: + r""" + How the TCP client reconnects after the connection to the server is lost. + """ + @property + def enabled(self) -> builtins.bool: ... + @property + def max_retries(self) -> builtins.int | None: ... + @property + def interval(self) -> datetime.timedelta: ... + @property + def reestablish_after(self) -> datetime.timedelta: ... + def __new__( + cls, + *, + enabled: builtins.bool = True, + max_retries: builtins.int | None = None, + interval: datetime.timedelta | None = None, + reestablish_after: datetime.timedelta | None = None, + ) -> TcpReconnectionConfig: + r""" + Constructs a reconnection policy, defaulting every unset field to the + value the Rust SDK uses. + + Args: + enabled: Whether to reconnect at all. + max_retries: Attempts before giving up, or `None` for unlimited. + interval: Delay between attempts. Defaults to 1 second. + reestablish_after: Cooldown before reconnecting after a previously + successful connection. Defaults to 5 seconds. + """ + def __repr__(self) -> builtins.str: ... + @typing.final class Topic: @property diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index b860f19b5d..9aeb8c4ec6 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -29,10 +29,12 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; use std::str::FromStr; use std::sync::Arc; +use crate::config::PyClientConfig; use crate::consumer::{ AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails, - IggyConsumer, py_delta_to_iggy_duration, + IggyConsumer, }; +use crate::duration::py_delta_to_iggy_duration; use crate::identifier::PyIdentifier; use crate::receive_message::{PollingStrategy, ReceiveMessage}; use crate::send_message::SendMessage; @@ -55,17 +57,41 @@ pub struct IggyClient { #[gen_stub_pymethods] #[pymethods] impl IggyClient { - /// Constructs a new IggyClient from a TCP server address. + /// Constructs a new IggyClient from a TCP server address or a `TcpConfig`. /// This initializes a new runtime for asynchronous operations. /// Future versions might utilize asyncio for more Pythonic async. + /// + /// Args: + /// conn: Either a `host:port` address, or a `TcpConfig` carrying the full + /// transport configuration. Defaults to `127.0.0.1:8090` with auto-login + /// disabled. + /// + /// Raises: + /// PyRuntimeError: If the address is not a valid `host:port` pair, or if the + /// client cannot be built. #[new] #[pyo3(signature = (conn=None))] fn new( - #[gen_stub(override_type(type_repr = "builtins.str | None"))] conn: Option, + #[gen_stub(override_type(type_repr = "TcpConfig | builtins.str | None"))] conn: Option< + PyClientConfig, + >, ) -> PyResult { + let config = match conn { + Some(PyClientConfig::Config(config)) => config.client_config(), + Some(PyClientConfig::ServerAddress(server_address)) => Arc::new( + TcpClientConfigBuilder::new() + .with_server_address(server_address) + .build() + .map_err(|e| { + PyErr::new::(e.to_string()) + })?, + ), + None => Arc::new(TcpClientConfig::default()), + }; + let tcp_client = TcpClient::create(config) + .map_err(|e| PyErr::new::(e.to_string()))?; let client = IggyClientBuilder::new() - .with_tcp() - .with_server_address(conn.unwrap_or("127.0.0.1:8090".to_string())) + .with_client(ClientWrapper::Tcp(tcp_client)) .build() .map_err(|e| PyErr::new::(e.to_string()))?; Ok(IggyClient { diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs new file mode 100644 index 0000000000..7bb1d651a5 --- /dev/null +++ b/foreign/python/src/config.rs @@ -0,0 +1,368 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy, defaulting every unset field to the + /// value the Rust SDK uses. + /// + /// Args: + /// enabled: Whether to reconnect at all. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + #[new] + #[pyo3(signature = (*, enabled=true, max_retries=None, interval=None, reestablish_after=None))] + fn new( + enabled: bool, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option>, + ) -> Self { + let defaults = RustTcpClientReconnectionConfig::default(); + Self { + inner: RustTcpClientReconnectionConfig { + enabled, + max_retries, + interval: interval + .as_ref() + .map_or(defaults.interval, py_delta_to_iggy_duration), + reestablish_after: reestablish_after + .as_ref() + .map_or(defaults.reestablish_after, py_delta_to_iggy_duration), + }, + } + } + + #[getter] + fn enabled(&self) -> bool { + self.inner.enabled + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn max_retries(&self) -> Option { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.interval) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.reestablish_after) + } + + fn __repr__(&self) -> String { + let max_retries = match self.inner.max_retries { + Some(max_retries) => max_retries.to_string(), + None => "None".to_owned(), + }; + format!( + "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + if self.inner.enabled { "True" } else { "False" }, + self.inner.interval.as_human_time_string(), + self.inner.reestablish_after.as_human_time_string(), + ) + } +} + +/// Configuration for the TCP transport, accepted by `IggyClient(...)`. +/// +/// Mirrors `TcpClientConfig` in the Rust SDK. Every field is keyword-only and +/// falls back to the same default the Rust SDK uses. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct TcpConfig { + auto_login: AutoLogin, + reconnection: TcpReconnectionConfig, + inner: Arc, +} + +impl TcpConfig { + /// The configuration in the shape `TcpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpConfig { + /// Constructs a TCP configuration, defaulting every unset field to the value + /// the Rust SDK uses. + /// + /// Args: + /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`. + /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + /// reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// tls_enabled: Whether to connect over TLS. + /// tls_domain: Domain to validate the certificate against. Empty means it is + /// taken from `server_address`. + /// tls_ca_file: Path to the CA file for TLS. + /// tls_validate_certificate: Whether to validate the server certificate. + /// nodelay: Disable the Nagle algorithm for the TCP socket. + /// + /// Raises: + /// PyValueError: If `server_address` is not a valid `host:port` pair. + #[new] + #[pyo3(signature = ( + *, + server_address=None, + auto_login=None, + reconnection=None, + heartbeat_interval=None, + tls_enabled=false, + tls_domain=None, + tls_ca_file=None, + tls_validate_certificate=true, + nodelay=false, + ))] + #[allow(clippy::too_many_arguments)] + fn new( + #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option< + String, + >, + #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option, + #[gen_stub(override_type(type_repr = "TcpReconnectionConfig | None"))] reconnection: Option< + TcpReconnectionConfig, + >, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option>, + tls_enabled: bool, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option, + tls_validate_certificate: bool, + nodelay: bool, + ) -> PyResult { + let defaults = RustTcpClientConfig::default(); + let auto_login = auto_login.unwrap_or_default(); + let reconnection = reconnection.unwrap_or_default(); + + let mut builder = TcpClientConfigBuilder::new() + .with_server_address(server_address.unwrap_or(defaults.server_address)) + .with_auto_sign_in(auto_login.inner.clone()) + .with_tls_enabled(tls_enabled) + .with_tls_domain(tls_domain.unwrap_or(defaults.tls_domain)) + .with_tls_validate_certificate(tls_validate_certificate); + if let Some(tls_ca_file) = tls_ca_file { + builder = builder.with_tls_ca_file(tls_ca_file); + } + if nodelay { + builder = builder.with_no_delay(); + } + + let mut inner = builder + .build() + .map_err(|e| PyErr::new::(e.to_string()))?; + // TcpClientConfigBuilder exposes no setter for either of these. + inner.heartbeat_interval = heartbeat_interval + .as_ref() + .map_or(defaults.heartbeat_interval, py_delta_to_iggy_duration); + inner.reconnection = reconnection.inner.clone(); + + Ok(Self { + auto_login, + reconnection, + inner: Arc::new(inner), + }) + } + + #[getter] + fn server_address(&self) -> String { + self.inner.server_address.clone() + } + + #[getter] + fn auto_login(&self) -> AutoLogin { + self.auto_login.clone() + } + + #[getter] + fn reconnection(&self) -> TcpReconnectionConfig { + self.reconnection.clone() + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.heartbeat_interval) + } + + #[getter] + fn tls_enabled(&self) -> bool { + self.inner.tls_enabled + } + + #[getter] + fn tls_domain(&self) -> String { + self.inner.tls_domain.clone() + } + + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn tls_ca_file(&self) -> Option { + self.inner.tls_ca_file.clone() + } + + #[getter] + fn tls_validate_certificate(&self) -> bool { + self.inner.tls_validate_certificate + } + + #[getter] + fn nodelay(&self) -> bool { + self.inner.nodelay + } + + fn __repr__(&self) -> String { + format!( + "TcpConfig(server_address={:?}, auto_login={}, reconnection={}, heartbeat_interval={}, tls_enabled={})", + self.inner.server_address, + self.auto_login.__repr__(), + self.reconnection.__repr__(), + self.inner.heartbeat_interval.as_human_time_string(), + if self.inner.tls_enabled { + "True" + } else { + "False" + }, + ) + } +} + +/// What `IggyClient(...)` accepts: a bare `host:port` or a full `TcpConfig`. +#[derive(FromPyObject)] +pub enum PyClientConfig { + #[pyo3(transparent)] + Config(TcpConfig), + #[pyo3(transparent, annotation = "str")] + ServerAddress(String), +} +impl_stub_type!(PyClientConfig = TcpConfig | String); diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 0b6066b686..41d849c79f 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs @@ -16,7 +16,6 @@ // under the License. use std::sync::Arc; -use std::time::Duration; use futures::StreamExt; use iggy::consumer_ext::{IggyConsumerMessageExt, MessageConsumer}; @@ -24,11 +23,11 @@ use iggy::prelude::{ AutoCommit as RustAutoCommit, AutoCommitAfter as RustAutoCommitAfter, AutoCommitWhen as RustAutoCommitWhen, ConsumerGroup as RustConsumerGroup, ConsumerGroupDetails as RustConsumerGroupDetails, - ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyDuration, - IggyError, ReceivedMessage, + ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyError, + ReceivedMessage, }; use pyo3::exceptions::PyStopAsyncIteration; -use pyo3::types::{PyDelta, PyDeltaAccess}; +use pyo3::types::PyDelta; use pyo3::prelude::*; use pyo3_async_runtimes::TaskLocals; @@ -39,6 +38,7 @@ use tokio::sync::Mutex; use tokio::sync::oneshot::Sender; use tokio::task::JoinHandle; +use crate::duration::py_delta_to_iggy_duration; use crate::identifier::PyIdentifier; use crate::receive_message::ReceiveMessage; @@ -516,12 +516,3 @@ impl PyStubType for AutoCommitAfter { TypeInfo::unqualified("AutoCommitAfter") } } - -pub fn py_delta_to_iggy_duration(delta1: &Py) -> IggyDuration { - Python::attach(|py| { - let delta = delta1.bind(py); - let seconds = (delta.get_days() * 60 * 60 * 24 + delta.get_seconds()) as u64; - let nanos = (delta.get_microseconds() * 1_000) as u32; - IggyDuration::new(Duration::new(seconds, nanos)) - }) -} diff --git a/foreign/python/src/duration.rs b/foreign/python/src/duration.rs new file mode 100644 index 0000000000..d0a807cfb5 --- /dev/null +++ b/foreign/python/src/duration.rs @@ -0,0 +1,43 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 iggy::prelude::IggyDuration; +use pyo3::prelude::*; +use pyo3::types::{PyDelta, PyDeltaAccess}; +use std::time::Duration; + +pub fn py_delta_to_iggy_duration(delta: &Py) -> IggyDuration { + Python::attach(|py| { + let delta = delta.bind(py); + let seconds = (delta.get_days() * 60 * 60 * 24 + delta.get_seconds()) as u64; + let nanos = (delta.get_microseconds() * 1_000) as u32; + IggyDuration::new(Duration::new(seconds, nanos)) + }) +} + +pub fn iggy_duration_to_py_delta( + py: Python<'_>, + duration: IggyDuration, +) -> PyResult> { + let micros = duration.as_micros(); + let seconds = i32::try_from(micros / 1_000_000).map_err(|_| { + PyErr::new::( + "duration does not fit into a datetime.timedelta", + ) + })?; + PyDelta::new(py, 0, seconds, (micros % 1_000_000) as i32, true) +} diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index f831fd703a..66ea8cec75 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -16,7 +16,9 @@ // under the License. pub mod client; +mod config; mod consumer; +mod duration; mod identifier; mod receive_message; mod send_message; @@ -25,6 +27,7 @@ mod topic; mod user; use client::IggyClient; +use config::{AutoLogin, TcpConfig, TcpReconnectionConfig}; use consumer::{ AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator, @@ -42,6 +45,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; From 4d5f97c0aed2a654d4237b1489f5f3e83c8e86f6 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:17:26 +0800 Subject: [PATCH 2/7] test(python): cover the TCP client configuration surface Round-trip every field through the getters so a default that drifts from the Rust SDK is caught, and assert that neither the password nor a personal access token comes back out of repr. The auto-login tests are the point of the configuration: a privileged call succeeds without a manual login_user() when credentials are configured, and fails without them. --- foreign/python/tests/test_client_config.py | 234 +++++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 foreign/python/tests/test_client_config.py diff --git a/foreign/python/tests/test_client_config.py b/foreign/python/tests/test_client_config.py new file mode 100644 index 0000000000..18c86f37c3 --- /dev/null +++ b/foreign/python/tests/test_client_config.py @@ -0,0 +1,234 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +""" +Tests for the TCP client configuration surface. + +`TcpConfig`, `TcpReconnectionConfig` and `AutoLogin` mirror the Rust SDK +types, so most of these assert that a value set from Python survives to the +getters and that unset fields fall back to the Rust defaults. The last class +proves the point of the configuration: with `auto_login` set, credentials are +replayed on connect and no manual `login_user()` is needed. +""" + +from datetime import timedelta + +import pytest + +from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig + +from .utils import get_server_config, wait_for_ping, wait_for_server + + +class TestAutoLogin: + """Test the credentials carried into the client.""" + + def test_disabled_has_no_username(self): + """Test that the disabled variant carries no credentials.""" + auto_login = AutoLogin.disabled() + + assert auto_login.enabled is False + assert auto_login.username is None + + def test_username_password_exposes_username_only(self): + """Test that the username is readable back but the password is not.""" + auto_login = AutoLogin.username_password("iggy", "secret") + + assert auto_login.enabled is True + assert auto_login.username == "iggy" + assert "secret" not in repr(auto_login) + + def test_personal_access_token_hides_the_token(self): + """Test that a token login exposes neither a username nor the token.""" + auto_login = AutoLogin.personal_access_token("secret-token") + + assert auto_login.enabled is True + assert auto_login.username is None + assert "secret-token" not in repr(auto_login) + + +class TestTcpReconnectionConfig: + """Test the reconnection policy.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured policy reconnects forever, one second apart.""" + reconnection = TcpReconnectionConfig() + + assert reconnection.enabled is True + assert reconnection.max_retries is None + assert reconnection.interval == timedelta(seconds=1) + assert reconnection.reestablish_after == timedelta(seconds=5) + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + reconnection = TcpReconnectionConfig( + enabled=False, + max_retries=10, + interval=timedelta(milliseconds=250), + reestablish_after=timedelta(seconds=30), + ) + + assert reconnection.enabled is False + assert reconnection.max_retries == 10 + assert reconnection.interval == timedelta(milliseconds=250) + assert reconnection.reestablish_after == timedelta(seconds=30) + + def test_arguments_are_keyword_only(self): + """Test that the adjacent flags cannot be passed positionally.""" + with pytest.raises(TypeError): + # pyrefly: ignore # bad-argument-count + TcpReconnectionConfig(True) + + +class TestTcpConfig: + """Test the transport configuration.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured transport matches the Rust SDK defaults.""" + config = TcpConfig() + + assert config.server_address == "127.0.0.1:8090" + assert config.auto_login.enabled is False + assert config.reconnection.enabled is True + assert config.heartbeat_interval == timedelta(seconds=5) + assert config.tls_enabled is False + assert config.tls_domain == "" + assert config.tls_ca_file is None + assert config.tls_validate_certificate is True + assert config.nodelay is False + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + config = TcpConfig( + server_address="localhost:8090", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=TcpReconnectionConfig(max_retries=3), + heartbeat_interval=timedelta(seconds=15), + tls_enabled=True, + tls_domain="localhost", + tls_ca_file="ca.pem", + tls_validate_certificate=False, + nodelay=True, + ) + + assert config.server_address == "localhost:8090" + assert config.auto_login.username == "iggy" + assert config.reconnection.max_retries == 3 + assert config.heartbeat_interval == timedelta(seconds=15) + assert config.tls_enabled is True + assert config.tls_domain == "localhost" + assert config.tls_ca_file == "ca.pem" + assert config.tls_validate_certificate is False + assert config.nodelay is True + + def test_arguments_are_keyword_only(self): + """Test that the address cannot be passed positionally.""" + with pytest.raises(TypeError): + # pyrefly: ignore # bad-argument-count + TcpConfig("127.0.0.1:8090") + + def test_repr_hides_the_password(self): + """Test that the password does not leak through repr.""" + config = TcpConfig(auto_login=AutoLogin.username_password("iggy", "secret")) + + assert "secret" not in repr(config) + + @pytest.mark.parametrize( + "invalid_address", + ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000", "::1:8090"], + ) + def test_invalid_server_address_is_rejected(self, invalid_address: str): + """Test that a malformed address fails at construction, not at connect.""" + with pytest.raises(ValueError): + TcpConfig(server_address=invalid_address) + + +class TestClientConstruction: + """Test what the client constructor accepts.""" + + def test_accepts_a_config(self): + """Test that a client can be built from a config object.""" + assert IggyClient(TcpConfig(server_address="127.0.0.1:8090")) is not None + + def test_accepts_an_address(self): + """Test that the address form still works.""" + assert IggyClient("127.0.0.1:8090") is not None + + def test_accepts_nothing(self): + """Test that the default address is used when no argument is given.""" + assert IggyClient() is not None + + def test_rejects_an_invalid_address(self): + """Test that a malformed address is rejected.""" + with pytest.raises(RuntimeError): + IggyClient("nonsense") + + +@pytest.mark.integration +class TestAutoLoginAgainstServer: + """Test that configured credentials are actually replayed on connect.""" + + @pytest.mark.asyncio + async def test_auto_login_authenticates_without_login_user(self, unique_name): + """Test that a privileged call succeeds without a manual login_user().""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient( + TcpConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "iggy"), + ) + ) + await client.connect() + await wait_for_ping(client) + + stream_name = unique_name() + await client.create_stream(stream_name) + assert await client.get_stream(stream_name) is not None + + @pytest.mark.asyncio + async def test_without_auto_login_a_privileged_call_is_unauthenticated( + self, unique_name + ): + """Test that the same call fails when no credentials are configured.""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient(TcpConfig(server_address=f"{host}:{port}")) + await client.connect() + await wait_for_ping(client) + + with pytest.raises(RuntimeError): + await client.create_stream(unique_name()) + + @pytest.mark.asyncio + async def test_wrong_auto_login_credentials_fail(self): + """Test that bad configured credentials surface as a connect failure.""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient( + TcpConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "invalid-password"), + reconnection=TcpReconnectionConfig(enabled=False), + ) + ) + + with pytest.raises(RuntimeError): + await client.connect() From 44e02e2cb2b3f376aabc9ca0b84926ff21758394 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:21:07 +0800 Subject: [PATCH 3/7] docs(python): add a client configuration example The existing examples all reach for a connection string, which leaves the new config types undiscoverable. This one configures auto-login and reconnection directly and never calls login_user, so the recovery the credentials unlock is visible: restart the server while it runs and the client picks up where it left off. --- examples/python/README.md | 14 ++ examples/python/client-configuration/main.py | 142 +++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 examples/python/client-configuration/main.py diff --git a/examples/python/README.md b/examples/python/README.md index 9bf943b75c..afe255b293 100644 --- a/examples/python/README.md +++ b/examples/python/README.md @@ -71,6 +71,20 @@ python basic/consumer.py Demonstrates fundamental client connection, authentication, batch message sending, and polling with support for TCP/QUIC/HTTP protocols. +### Client Configuration + +Auto-login and reconnection, configured explicitly rather than through a connection string: + +```bash +# Using uv +uv run client-configuration/main.py + +# Without using uv +python client-configuration/main.py +``` + +Demonstrates `TcpConfig`, `TcpReconnectionConfig` and `AutoLogin`. Because the credentials are replayed on every connect, the client recovers its session after the server restarts instead of failing with `Unauthenticated`. + ## TLS Examples To test with a TLS-enabled server, start the server with TLS configured (see main README), then run: diff --git a/examples/python/client-configuration/main.py b/examples/python/client-configuration/main.py new file mode 100644 index 0000000000..22f07762a4 --- /dev/null +++ b/examples/python/client-configuration/main.py @@ -0,0 +1,142 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +""" +Configures the TCP client explicitly instead of passing a bare address. + +The point of `auto_login` is that the credentials are replayed every time the +client connects, including after a reconnect. That is what lets the SDK +recover a session the server dropped: without it, a restart of the server +surfaces as `Unauthenticated` on the next call and the application has to +reconnect and log in by hand. + +Run this, then restart the server while it is polling: the client reconnects, +replays the login and keeps going. +""" + +import argparse +import asyncio +import typing +from datetime import timedelta + +from apache_iggy import ( + AutoLogin, + IggyClient, + PollingStrategy, + StreamDetails, + TcpConfig, + TcpReconnectionConfig, + TopicDetails, +) +from apache_iggy import SendMessage as Message +from loguru import logger + +STREAM_NAME = "configured-stream" +TOPIC_NAME = "configured-topic" +PARTITION_ID = 0 +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + tcp_server_address: str + username: str + password: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--tcp-server-address", + default="127.0.0.1:8090", + help="Iggy TCP server address (host:port)", + ) + parser.add_argument("--username", default="iggy", help="Username to log in with") + parser.add_argument("--password", default="iggy", help="Password to log in with") + return ArgNamespace(**vars(parser.parse_args())) + + +def build_config(args: ArgNamespace) -> TcpConfig: + return TcpConfig( + server_address=args.tcp_server_address, + auto_login=AutoLogin.username_password(args.username, args.password), + reconnection=TcpReconnectionConfig( + enabled=True, + max_retries=None, # retry forever + interval=timedelta(seconds=1), + reestablish_after=timedelta(seconds=5), + ), + heartbeat_interval=timedelta(seconds=5), + nodelay=True, + ) + + +async def main(): + args = parse_args() + config = build_config(args) + logger.info(f"Connecting with {config}") + + client = IggyClient(config) + # No login_user() call: auto_login replays the credentials on every connect. + await client.connect() + logger.info("Connected and authenticated.") + + await init_system(client) + await produce_and_consume(client) + + +async def init_system(client: IggyClient): + stream: StreamDetails | None = await client.get_stream(STREAM_NAME) + if stream is None: + await client.create_stream(name=STREAM_NAME) + logger.info(f"Created stream {STREAM_NAME}.") + + topic: TopicDetails | None = await client.get_topic(STREAM_NAME, TOPIC_NAME) + if topic is None: + await client.create_topic( + stream=STREAM_NAME, + name=TOPIC_NAME, + partitions_count=1, + replication_factor=1, + ) + logger.info(f"Created topic {TOPIC_NAME}.") + + +async def produce_and_consume(client: IggyClient): + for batch in range(BATCHES_LIMIT): + messages = [Message(f"message-{batch}-{i}") for i in range(10)] + await client.send_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + partitioning=PARTITION_ID, + messages=messages, + ) + logger.info(f"Sent batch {batch}.") + + polled = await client.poll_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.Next(), + count=len(messages), + auto_commit=True, + ) + logger.info(f"Polled {len(polled)} messages.") + await asyncio.sleep(0.5) + + +if __name__ == "__main__": + asyncio.run(main()) From 76ab557799dcab497055c66094dce9ff138c2975 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:25:54 +0800 Subject: [PATCH 4/7] docs(python): document client configuration in the SDK README The README pointed only at the examples directory, so the configuration surface stayed invisible to anyone reading the package page on PyPI. --- foreign/python/README.md | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/foreign/python/README.md b/foreign/python/README.md index f01754f6dc..898ef9d5dd 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -58,6 +58,38 @@ maturin develop pytest tests/ -v # Run tests (requires iggy-server running) ``` +## Client Configuration + +`IggyClient` takes either a server address or a `TcpConfig`. Configuring `auto_login` +lets the SDK replay the credentials whenever it reconnects, so a session dropped by a +server restart is recovered instead of surfacing as `Unauthenticated`: + +```python +from datetime import timedelta + +from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig + +client = IggyClient( + TcpConfig( + server_address="127.0.0.1:8090", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=TcpReconnectionConfig( + enabled=True, + max_retries=10, + interval=timedelta(seconds=2), + reestablish_after=timedelta(seconds=30), + ), + heartbeat_interval=timedelta(seconds=5), + ) +) +await client.connect() +``` + +`TcpConfig` also carries `tls_enabled`, `tls_domain`, `tls_ca_file`, +`tls_validate_certificate` and `nodelay`. Every field is keyword-only and defaults to the +same value the Rust SDK uses. `IggyClient.from_connection_string(...)` remains available +for the same settings in string form. + ## Examples Refer to the [examples/python/](https://github.com/apache/iggy/tree/master/examples/python) directory for usage examples. From 921cd8f3965e9a421e99ce61b1af19c480068b76 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:57:05 +0800 Subject: [PATCH 5/7] fix(python): validate durations and derive config defaults from Rust A negative timedelta normalizes to negative days plus positive seconds, so the old conversion summed to a negative i32 and cast it to u64, turning interval=timedelta(seconds=-1) into u64::MAX seconds: the config constructed fine and the client then slept forever on reconnect. Days arithmetic also overflowed i32 beyond ~68 years, and the reverse conversion stuffed everything into the seconds argument so such values could not read back. Conversion is now fallible, rejects negative input with ValueError at construction, computes in i64, and splits days on the way out. The AutoCommit conversion becomes TryFrom to carry the error. The boolean constructor defaults were literals in the pyo3 signature, so a change to a Rust default would silently not propagate. They are now Option arguments that fall back to TcpClientConfig::default(), the same way the durations already did. --- foreign/python/apache_iggy.pyi | 22 ++++++---- foreign/python/src/client.rs | 16 +++---- foreign/python/src/config.rs | 79 +++++++++++++++++++--------------- foreign/python/src/consumer.rs | 19 ++++---- foreign/python/src/duration.rs | 19 +++++--- 5 files changed, 90 insertions(+), 65 deletions(-) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 3befd0deb7..c503a35f78 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -996,11 +996,11 @@ class TcpConfig: auto_login: AutoLogin | None = None, reconnection: TcpReconnectionConfig | None = None, heartbeat_interval: datetime.timedelta | None = None, - tls_enabled: builtins.bool = False, + tls_enabled: builtins.bool | None = None, tls_domain: builtins.str | None = None, tls_ca_file: builtins.str | None = None, - tls_validate_certificate: builtins.bool = True, - nodelay: builtins.bool = False, + tls_validate_certificate: builtins.bool | None = None, + nodelay: builtins.bool | None = None, ) -> TcpConfig: r""" Constructs a TCP configuration, defaulting every unset field to the value @@ -1011,15 +1011,18 @@ class TcpConfig: auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. - tls_enabled: Whether to connect over TLS. + tls_enabled: Whether to connect over TLS. Defaults to disabled. tls_domain: Domain to validate the certificate against. Empty means it is taken from `server_address`. tls_ca_file: Path to the CA file for TLS. tls_validate_certificate: Whether to validate the server certificate. - nodelay: Disable the Nagle algorithm for the TCP socket. + Defaults to validating. + nodelay: Disable the Nagle algorithm for the TCP socket. Defaults to + leaving it on. Raises: - PyValueError: If `server_address` is not a valid `host:port` pair. + PyValueError: If `server_address` is not a valid `host:port` pair, or + if a duration is negative. """ def __repr__(self) -> builtins.str: ... @@ -1039,7 +1042,7 @@ class TcpReconnectionConfig: def __new__( cls, *, - enabled: builtins.bool = True, + enabled: builtins.bool | None = None, max_retries: builtins.int | None = None, interval: datetime.timedelta | None = None, reestablish_after: datetime.timedelta | None = None, @@ -1049,11 +1052,14 @@ class TcpReconnectionConfig: value the Rust SDK uses. Args: - enabled: Whether to reconnect at all. + enabled: Whether to reconnect at all. Defaults to enabled. max_retries: Attempts before giving up, or `None` for unlimited. interval: Delay between attempts. Defaults to 1 second. reestablish_after: Cooldown before reconnecting after a previously successful connection. Defaults to 5 seconds. + + Raises: + PyValueError: If a duration is negative. """ def __repr__(self) -> builtins.str: ... diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 9aeb8c4ec6..c7dc46974a 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -17,8 +17,8 @@ use bytes::Bytes; use iggy::prelude::{ - Consumer as RustConsumer, IggyClient as RustIggyClient, IggyMessage as RustMessage, - PollingStrategy as RustPollingStrategy, *, + AutoCommit as RustAutoCommit, Consumer as RustConsumer, IggyClient as RustIggyClient, + IggyMessage as RustMessage, PollingStrategy as RustPollingStrategy, *, }; use pyo3::PyRef; use pyo3::prelude::*; @@ -369,7 +369,7 @@ impl IggyClient { }; let expiry = match message_expiry { - Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)), + Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)?), None => IggyExpiry::ServerDefault, }; @@ -492,7 +492,7 @@ impl IggyClient { }; let expiry = match message_expiry { - Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)), + Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)?), None => IggyExpiry::ServerDefault, }; @@ -951,16 +951,16 @@ impl IggyClient { builder = builder.batch_length(batch_length) }; if let Some(auto_commit) = auto_commit { - builder = builder.auto_commit(auto_commit.into()) + builder = builder.auto_commit(RustAutoCommit::try_from(auto_commit)?) }; if let Some(poll_interval) = poll_interval { - builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval)) + builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval)?) } else { builder = builder.without_poll_interval() }; if let Some(polling_retry_interval) = polling_retry_interval { builder = - builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval)) + builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval)?) } if init_retries.is_some() && init_retry_interval.is_none() { return Err(PyErr::new::( @@ -976,7 +976,7 @@ impl IggyClient { { builder = builder.init_retries( init_retries, - py_delta_to_iggy_duration(&init_retry_interval), + py_delta_to_iggy_duration(&init_retry_interval)?, ); } if allow_replay { diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs index 7bb1d651a5..209ed6bb58 100644 --- a/foreign/python/src/config.rs +++ b/foreign/python/src/config.rs @@ -124,34 +124,41 @@ impl TcpReconnectionConfig { /// value the Rust SDK uses. /// /// Args: - /// enabled: Whether to reconnect at all. + /// enabled: Whether to reconnect at all. Defaults to enabled. /// max_retries: Attempts before giving up, or `None` for unlimited. /// interval: Delay between attempts. Defaults to 1 second. /// reestablish_after: Cooldown before reconnecting after a previously /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// PyValueError: If a duration is negative. #[new] - #[pyo3(signature = (*, enabled=true, max_retries=None, interval=None, reestablish_after=None))] + #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))] fn new( - enabled: bool, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] enabled: Option, #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option, #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] interval: Option>, #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] reestablish_after: Option>, - ) -> Self { + ) -> PyResult { let defaults = RustTcpClientReconnectionConfig::default(); - Self { + Ok(Self { inner: RustTcpClientReconnectionConfig { - enabled, + enabled: enabled.unwrap_or(defaults.enabled), max_retries, interval: interval .as_ref() - .map_or(defaults.interval, py_delta_to_iggy_duration), + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.interval), reestablish_after: reestablish_after .as_ref() - .map_or(defaults.reestablish_after, py_delta_to_iggy_duration), + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.reestablish_after), }, - } + }) } #[getter] @@ -222,15 +229,18 @@ impl TcpConfig { /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. /// reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. - /// tls_enabled: Whether to connect over TLS. + /// tls_enabled: Whether to connect over TLS. Defaults to disabled. /// tls_domain: Domain to validate the certificate against. Empty means it is /// taken from `server_address`. /// tls_ca_file: Path to the CA file for TLS. /// tls_validate_certificate: Whether to validate the server certificate. - /// nodelay: Disable the Nagle algorithm for the TCP socket. + /// Defaults to validating. + /// nodelay: Disable the Nagle algorithm for the TCP socket. Defaults to + /// leaving it on. /// /// Raises: - /// PyValueError: If `server_address` is not a valid `host:port` pair. + /// PyValueError: If `server_address` is not a valid `host:port` pair, or + /// if a duration is negative. #[new] #[pyo3(signature = ( *, @@ -238,11 +248,11 @@ impl TcpConfig { auto_login=None, reconnection=None, heartbeat_interval=None, - tls_enabled=false, + tls_enabled=None, tls_domain=None, tls_ca_file=None, - tls_validate_certificate=true, - nodelay=false, + tls_validate_certificate=None, + nodelay=None, ))] #[allow(clippy::too_many_arguments)] fn new( @@ -255,37 +265,38 @@ impl TcpConfig { >, #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] heartbeat_interval: Option>, - tls_enabled: bool, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option, #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option, #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option, - tls_validate_certificate: bool, - nodelay: bool, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + tls_validate_certificate: Option, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] nodelay: Option, ) -> PyResult { let defaults = RustTcpClientConfig::default(); let auto_login = auto_login.unwrap_or_default(); let reconnection = reconnection.unwrap_or_default(); - let mut builder = TcpClientConfigBuilder::new() + // The builder is only used to validate and trim the server address; the + // remaining fields are assigned directly so every unset argument falls + // back to the Rust `TcpClientConfig::default()` value instead of a + // literal duplicated here. + let mut inner = TcpClientConfigBuilder::new() .with_server_address(server_address.unwrap_or(defaults.server_address)) - .with_auto_sign_in(auto_login.inner.clone()) - .with_tls_enabled(tls_enabled) - .with_tls_domain(tls_domain.unwrap_or(defaults.tls_domain)) - .with_tls_validate_certificate(tls_validate_certificate); - if let Some(tls_ca_file) = tls_ca_file { - builder = builder.with_tls_ca_file(tls_ca_file); - } - if nodelay { - builder = builder.with_no_delay(); - } - - let mut inner = builder .build() .map_err(|e| PyErr::new::(e.to_string()))?; - // TcpClientConfigBuilder exposes no setter for either of these. + inner.auto_login = auto_login.inner.clone(); + inner.reconnection = reconnection.inner.clone(); inner.heartbeat_interval = heartbeat_interval .as_ref() - .map_or(defaults.heartbeat_interval, py_delta_to_iggy_duration); - inner.reconnection = reconnection.inner.clone(); + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.heartbeat_interval); + inner.tls_enabled = tls_enabled.unwrap_or(defaults.tls_enabled); + inner.tls_domain = tls_domain.unwrap_or(defaults.tls_domain); + inner.tls_ca_file = tls_ca_file.or(defaults.tls_ca_file); + inner.tls_validate_certificate = + tls_validate_certificate.unwrap_or(defaults.tls_validate_certificate); + inner.nodelay = nodelay.unwrap_or(defaults.nodelay); Ok(Self { auto_login, diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 41d849c79f..fb95b1d0bf 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs @@ -429,25 +429,24 @@ pub enum AutoCommit { After(AutoCommitAfter), } -impl From<&AutoCommit> for RustAutoCommit { - fn from(val: &AutoCommit) -> RustAutoCommit { - match val { +impl TryFrom<&AutoCommit> for RustAutoCommit { + type Error = PyErr; + + fn try_from(val: &AutoCommit) -> PyResult { + Ok(match val { AutoCommit::Disabled() => RustAutoCommit::Disabled, AutoCommit::Interval(delta) => { - let duration = py_delta_to_iggy_duration(delta); - RustAutoCommit::Interval(duration) + RustAutoCommit::Interval(py_delta_to_iggy_duration(delta)?) } AutoCommit::IntervalOrWhen(delta, when) => { - let duration = py_delta_to_iggy_duration(delta); - RustAutoCommit::IntervalOrWhen(duration, when.into()) + RustAutoCommit::IntervalOrWhen(py_delta_to_iggy_duration(delta)?, when.into()) } AutoCommit::IntervalOrAfter(delta, after) => { - let duration = py_delta_to_iggy_duration(delta); - RustAutoCommit::IntervalOrAfter(duration, after.into()) + RustAutoCommit::IntervalOrAfter(py_delta_to_iggy_duration(delta)?, after.into()) } AutoCommit::When(when) => RustAutoCommit::When(when.into()), AutoCommit::After(after) => RustAutoCommit::After(after.into()), - } + }) } } diff --git a/foreign/python/src/duration.rs b/foreign/python/src/duration.rs index d0a807cfb5..03126d0676 100644 --- a/foreign/python/src/duration.rs +++ b/foreign/python/src/duration.rs @@ -20,12 +20,19 @@ use pyo3::prelude::*; use pyo3::types::{PyDelta, PyDeltaAccess}; use std::time::Duration; -pub fn py_delta_to_iggy_duration(delta: &Py) -> IggyDuration { +pub fn py_delta_to_iggy_duration(delta: &Py) -> PyResult { Python::attach(|py| { let delta = delta.bind(py); - let seconds = (delta.get_days() * 60 * 60 * 24 + delta.get_seconds()) as u64; + // Python normalizes a negative timedelta to negative days plus + // non-negative seconds/microseconds, so the sign lives in the sum. + let seconds = i64::from(delta.get_days()) * 60 * 60 * 24 + i64::from(delta.get_seconds()); + if seconds < 0 { + return Err(PyErr::new::( + "duration must not be negative", + )); + } let nanos = (delta.get_microseconds() * 1_000) as u32; - IggyDuration::new(Duration::new(seconds, nanos)) + Ok(IggyDuration::new(Duration::new(seconds as u64, nanos))) }) } @@ -34,10 +41,12 @@ pub fn iggy_duration_to_py_delta( duration: IggyDuration, ) -> PyResult> { let micros = duration.as_micros(); - let seconds = i32::try_from(micros / 1_000_000).map_err(|_| { + let total_seconds = micros / 1_000_000; + let days = i32::try_from(total_seconds / 86_400).map_err(|_| { PyErr::new::( "duration does not fit into a datetime.timedelta", ) })?; - PyDelta::new(py, 0, seconds, (micros % 1_000_000) as i32, true) + let seconds = (total_seconds % 86_400) as i32; + PyDelta::new(py, days, seconds, (micros % 1_000_000) as i32, true) } From a82351c10b17e2cc3e98f1b666c5c5e02cabbf61 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:57:15 +0800 Subject: [PATCH 6/7] test(python): make the unit marker selectable and pin duration edges conftest auto-marked every module as integration, so tests explicitly marked unit could not be selected with -m "not integration" even though they need no server. The auto-mark now skips them. New cases pin the duration boundaries (negative rejected, zero legal, beyond the i32 seconds range round-trips) and the README claim that a connection string and TcpConfig reach the same behavior. --- foreign/python/tests/conftest.py | 5 ++ foreign/python/tests/test_client_config.py | 69 ++++++++++++++++++++++ 2 files changed, 74 insertions(+) diff --git a/foreign/python/tests/conftest.py b/foreign/python/tests/conftest.py index aa54ff50d8..3ab97065f5 100644 --- a/foreign/python/tests/conftest.py +++ b/foreign/python/tests/conftest.py @@ -131,5 +131,10 @@ def pytest_collection_modifyitems(items): path.name for path in Path(__file__).parent.glob("test_*.py") } for item in items: + # Tests explicitly marked as unit need no server; auto-marking them + # integration too would make `-m "not integration"` unable to select + # them. + if item.get_closest_marker("unit"): + continue if any(module in item.nodeid for module in integration_modules): item.add_marker(pytest.mark.integration) diff --git a/foreign/python/tests/test_client_config.py b/foreign/python/tests/test_client_config.py index 18c86f37c3..50a0f7a882 100644 --- a/foreign/python/tests/test_client_config.py +++ b/foreign/python/tests/test_client_config.py @@ -25,6 +25,7 @@ replayed on connect and no manual `login_user()` is needed. """ +from collections.abc import Callable from datetime import timedelta import pytest @@ -34,6 +35,7 @@ from .utils import get_server_config, wait_for_ping, wait_for_server +@pytest.mark.unit class TestAutoLogin: """Test the credentials carried into the client.""" @@ -61,6 +63,7 @@ def test_personal_access_token_hides_the_token(self): assert "secret-token" not in repr(auto_login) +@pytest.mark.unit class TestTcpReconnectionConfig: """Test the reconnection policy.""" @@ -93,7 +96,41 @@ def test_arguments_are_keyword_only(self): # pyrefly: ignore # bad-argument-count TcpReconnectionConfig(True) + @pytest.mark.parametrize( + "construct", + [ + lambda duration: TcpReconnectionConfig(interval=duration), + lambda duration: TcpReconnectionConfig(reestablish_after=duration), + ], + ids=["interval", "reestablish_after"], + ) + @pytest.mark.parametrize( + "negative", + [timedelta(microseconds=-1), timedelta(seconds=-1), timedelta(days=-1)], + ) + def test_negative_duration_is_rejected( + self, + construct: Callable[[timedelta], TcpReconnectionConfig], + negative: timedelta, + ): + """Test that a negative duration fails at construction, not at connect.""" + with pytest.raises(ValueError, match="negative"): + construct(negative) + + def test_zero_interval_is_allowed(self): + """Test that a zero interval is legal and readable back.""" + reconnection = TcpReconnectionConfig(interval=timedelta(0)) + + assert reconnection.interval == timedelta(0) + def test_very_long_interval_round_trips(self): + """Test that an interval beyond 68 years survives the i32 boundary.""" + reconnection = TcpReconnectionConfig(interval=timedelta(days=30_000)) + + assert reconnection.interval == timedelta(days=30_000) + + +@pytest.mark.unit class TestTcpConfig: """Test the transport configuration.""" @@ -156,7 +193,13 @@ def test_invalid_server_address_is_rejected(self, invalid_address: str): with pytest.raises(ValueError): TcpConfig(server_address=invalid_address) + def test_negative_heartbeat_interval_is_rejected(self): + """Test that a negative heartbeat interval fails at construction.""" + with pytest.raises(ValueError, match="negative"): + TcpConfig(heartbeat_interval=timedelta(seconds=-3)) + +@pytest.mark.unit class TestClientConstruction: """Test what the client constructor accepts.""" @@ -216,6 +259,32 @@ async def test_without_auto_login_a_privileged_call_is_unauthenticated( with pytest.raises(RuntimeError): await client.create_stream(unique_name()) + @pytest.mark.asyncio + async def test_config_and_connection_string_are_equivalent(self, unique_name): + """Test that TcpConfig and a connection string reach the same behavior.""" + host, port = get_server_config() + wait_for_server(host, port) + + from_config = IggyClient( + TcpConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=TcpReconnectionConfig( + max_retries=3, interval=timedelta(seconds=1) + ), + ) + ) + from_string = IggyClient.from_connection_string( + f"iggy+tcp://iggy:iggy@{host}:{port}" + "?reconnection_retries=3&reconnection_interval=1s" + ) + + stream_name = unique_name() + for client in (from_config, from_string): + await client.connect() + await wait_for_ping(client) + assert await client.get_stream(stream_name) is None + @pytest.mark.asyncio async def test_wrong_auto_login_credentials_fail(self): """Test that bad configured credentials surface as a connect failure.""" From b3190f4a1a7e499707201edeb31d8b6882115499 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Tue, 28 Jul 2026 21:57:15 +0800 Subject: [PATCH 7/7] docs(python): make the README configuration snippet runnable The snippet ended with a top-level await; every other sample in the repo wraps in asyncio.run, so paste-and-run failed on the only snippet a PyPI reader sees first. --- foreign/python/README.md | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/foreign/python/README.md b/foreign/python/README.md index 898ef9d5dd..74149f9fa3 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -65,24 +65,30 @@ lets the SDK replay the credentials whenever it reconnects, so a session dropped server restart is recovered instead of surfacing as `Unauthenticated`: ```python +import asyncio from datetime import timedelta from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig -client = IggyClient( - TcpConfig( - server_address="127.0.0.1:8090", - auto_login=AutoLogin.username_password("iggy", "iggy"), - reconnection=TcpReconnectionConfig( - enabled=True, - max_retries=10, - interval=timedelta(seconds=2), - reestablish_after=timedelta(seconds=30), - ), - heartbeat_interval=timedelta(seconds=5), + +async def main(): + client = IggyClient( + TcpConfig( + server_address="127.0.0.1:8090", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=TcpReconnectionConfig( + enabled=True, + max_retries=10, + interval=timedelta(seconds=2), + reestablish_after=timedelta(seconds=30), + ), + heartbeat_interval=timedelta(seconds=5), + ) ) -) -await client.connect() + await client.connect() + + +asyncio.run(main()) ``` `TcpConfig` also carries `tls_enabled`, `tls_domain`, `tls_ca_file`,