From c5aa04dec3e461ec31d90d4a52d4f539bf0203db Mon Sep 17 00:00:00 2001 From: saie-ch <132209179+saie-ch@users.noreply.github.com> Date: Sat, 29 Aug 2026 20:47:04 +0530 Subject: [PATCH 1/3] feat(python): add HttpConfig transport configuration Relates to #2835. --- .../python-maturin/pre-merge/action.yml | 1 + .github/workflows/coverage-baseline.yml | 1 + core/sdk/src/prelude.rs | 1 + examples/python/README.md | 13 ++ examples/python/http/consumer.py | 156 ++++++++++++++++ examples/python/http/producer.py | 166 ++++++++++++++++++ foreign/python/README.md | 25 +++ foreign/python/apache_iggy.pyi | 58 ++++++ foreign/python/src/client.rs | 26 ++- foreign/python/src/config.rs | 105 +++++++++++ foreign/python/src/lib.rs | 3 +- foreign/python/tests/test_http_config.py | 145 +++++++++++++++ foreign/python/tests/utils.py | 30 +++- 13 files changed, 725 insertions(+), 5 deletions(-) create mode 100644 examples/python/http/consumer.py create mode 100644 examples/python/http/producer.py create mode 100644 foreign/python/tests/test_http_config.py diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml index f97462f34d..1a313f180c 100644 --- a/.github/actions/python-maturin/pre-merge/action.yml +++ b/.github/actions/python-maturin/pre-merge/action.yml @@ -153,6 +153,7 @@ runs: # overwrite the coverage-instrumented .so with a non-instrumented one IGGY_SERVER_HOST=127.0.0.1 \ IGGY_SERVER_TCP_PORT=8090 \ + IGGY_SERVER_HTTP_PORT=3000 \ IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \ uv run --no-sync pytest tests/ -v \ --junitxml=../../reports/python-junit.xml \ diff --git a/.github/workflows/coverage-baseline.yml b/.github/workflows/coverage-baseline.yml index baa1ae160c..5a5d824bda 100644 --- a/.github/workflows/coverage-baseline.yml +++ b/.github/workflows/coverage-baseline.yml @@ -349,6 +349,7 @@ jobs: cd foreign/python IGGY_SERVER_HOST=127.0.0.1 \ IGGY_SERVER_TCP_PORT=8090 \ + IGGY_SERVER_HTTP_PORT=3000 \ IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \ uv run --no-sync pytest tests/ -v \ --junitxml=../../reports/python-junit.xml \ diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index 81f7d8c16d..59456858ee 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -41,6 +41,7 @@ pub use crate::clients::producer_builder::IggyProducerBuilder; pub use crate::clients::producer_config::{BackgroundConfig, DirectConfig}; pub use crate::clients::producer_sharding::{BalancedSharding, OrderedSharding, Sharding}; pub use crate::consumer_ext::IggyConsumerMessageExt; +pub use crate::http::http_client::HttpClient; pub use crate::stream_builder::IggyConsumerConfig; pub use crate::stream_builder::IggyStreamConsumer; pub use crate::stream_builder::{IggyProducerConfig, IggyStreamProducer}; diff --git a/examples/python/README.md b/examples/python/README.md index 7e5da180d4..7e029cd185 100644 --- a/examples/python/README.md +++ b/examples/python/README.md @@ -94,6 +94,19 @@ python message-headers/typed-headers/producer.py python message-headers/typed-headers/consumer.py ``` +## Transport Protocol Examples + +### HTTP + +Uses the explicit `IggyClient.http()` constructor. Assumes a server started with defaults, +which enables all four transports (`cargo run --bin iggy-server`, or the `docker run` command +above). + +```bash +uv run http/producer.py +uv run http/consumer.py +``` + ## 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/http/consumer.py b/examples/python/http/consumer.py new file mode 100644 index 0000000000..ef5bd73ba1 --- /dev/null +++ b/examples/python/http/consumer.py @@ -0,0 +1,156 @@ +# 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. + +import argparse +import asyncio +import typing +import urllib.parse + +from apache_iggy import ( + Consumer, + HttpConfig, + IggyClient, + PollingStrategy, + ReceiveMessage, +) +from loguru import logger + +STREAM_NAME = "sample-stream" +TOPIC_NAME = "sample-topic" +STREAM_ID = 0 +TOPIC_ID = 0 +PARTITION_ID = 0 +CONSUMER_NAME = "sample-consumer" +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + api_url: str + retries: int + + +class ValidateUrl(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str, + _option_string: str | None = None, + ): + parsed_url: urllib.parse.ParseResult = urllib.parse.urlparse(values) + if parsed_url.scheme not in ("http", "https") or parsed_url.netloc == "": + parser.error(f"Invalid API URL: {values}") + setattr(namespace, self.dest, values) + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--api-url", + help="Iggy HTTP API URL", + action=ValidateUrl, + default="http://127.0.0.1:3000", + ) + parser.add_argument( + "--retries", + type=int, + default=3, + help="Number of retries to perform on transient errors", + ) + args = parser.parse_args() + return ArgNamespace(**vars(args)) + + +def build_config(args: ArgNamespace) -> HttpConfig: + """Build an HTTP client configuration.""" + + return HttpConfig( + api_url=args.api_url, + retries=args.retries, + ) + + +async def main(): + args: ArgNamespace = parse_args() + try: + config = build_config(args) + except ValueError as error: + logger.error(f"Invalid client configuration: {error}") + return + logger.info(f"Connecting to {args.api_url}") + + client = IggyClient.http(config) + try: + logger.info("Connecting to IggyClient...") + await client.connect() + logger.info("Connected.") + # HTTP is a stateless transport: log in explicitly rather than relying + # on auto-login, which HttpConfig does not expose. + await client.login_user("iggy", "iggy") + await consume_messages(client) + except Exception as error: + logger.exception(f"Exception occurred in main function: {error}") + + +async def consume_messages(client: IggyClient): + interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep + logger.info( + f"Messages will be consumed from stream: {STREAM_NAME}, " + f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} " + f"with interval {interval * 1000} ms." + ) + offset = 0 + messages_per_batch = 10 + n_consumed_batches = 0 + while n_consumed_batches < BATCHES_LIMIT: + try: + logger.debug("Polling for messages...") + polled_messages = await client.poll_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + consumer=Consumer.Single(CONSUMER_NAME), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.Next(), + count=messages_per_batch, + auto_commit=True, + ) + if not polled_messages: + logger.info("No messages found in current poll") + await asyncio.sleep(interval) + continue + + offset += len(polled_messages) + for message in polled_messages: + handle_message(message) + n_consumed_batches += 1 + await asyncio.sleep(interval) + except Exception as error: + logger.exception(f"Exception occurred while consuming messages: {error}") + break + + logger.info(f"Consumed {n_consumed_batches} batches of messages, exiting.") + + +def handle_message(message: ReceiveMessage): + payload = message.payload().decode("utf-8") + logger.info( + f"Handling message at offset: {message.offset()} with payload: {payload}..." + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/python/http/producer.py b/examples/python/http/producer.py new file mode 100644 index 0000000000..7355e76421 --- /dev/null +++ b/examples/python/http/producer.py @@ -0,0 +1,166 @@ +# 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. + +import argparse +import asyncio +import typing +import urllib.parse + +from apache_iggy import HttpConfig, IggyClient, StreamDetails, TopicDetails +from apache_iggy import SendMessage as Message +from loguru import logger + +STREAM_NAME = "sample-stream" +TOPIC_NAME = "sample-topic" +STREAM_ID = 0 +TOPIC_ID = 0 +PARTITION_ID = 0 +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + api_url: str + retries: int + + +class ValidateUrl(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str, + _option_string: str | None = None, + ): + parsed_url: urllib.parse.ParseResult = urllib.parse.urlparse(values) + if parsed_url.scheme not in ("http", "https") or parsed_url.netloc == "": + parser.error(f"Invalid API URL: {values}") + setattr(namespace, self.dest, values) + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--api-url", + help="Iggy HTTP API URL", + action=ValidateUrl, + default="http://127.0.0.1:3000", + ) + parser.add_argument( + "--retries", + type=int, + default=3, + help="Number of retries to perform on transient errors", + ) + args = parser.parse_args() + return ArgNamespace(**vars(args)) + + +def build_config(args: ArgNamespace) -> HttpConfig: + """Build an HTTP client configuration.""" + + return HttpConfig( + api_url=args.api_url, + retries=args.retries, + ) + + +async def main(): + args: ArgNamespace = parse_args() + try: + config = build_config(args) + except ValueError as error: + logger.error(f"Invalid client configuration: {error}") + return + logger.info(f"Connecting to {args.api_url}") + + client = IggyClient.http(config) + logger.info("Connecting to IggyClient") + await client.connect() + logger.info("Connected.") + # HTTP is a stateless transport: log in explicitly rather than relying + # on auto-login, which HttpConfig does not expose. + await client.login_user("iggy", "iggy") + await init_system(client) + await produce_messages(client) + + +async def init_system(client: IggyClient): + logger.info(f"Creating stream with name {STREAM_NAME}...") + stream: StreamDetails | None = await client.get_stream(STREAM_NAME) + if stream is None: + await client.create_stream(name=STREAM_NAME) + logger.info("Stream was created successfully.") + else: + logger.warning(f"Stream {stream.name} already exists with ID {stream.id}") + + logger.info(f"Creating topic {TOPIC_NAME} in 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, + partitions_count=1, + name=TOPIC_NAME, + ) + logger.info("Topic was created successfully.") + else: + logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + + +async def produce_messages(client: IggyClient): + interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep + logger.info( + f"Messages will be sent to stream: {STREAM_NAME}, " + f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} " + f"with interval {interval * 1000} ms." + ) + current_id = 0 + messages_per_batch = 10 + n_sent_batches = 0 + while n_sent_batches < BATCHES_LIMIT: + messages = [] + for _ in range(messages_per_batch): + current_id += 1 + payload = f"message-{current_id}" + message = Message(payload) + messages.append(message) + logger.info( + f"Attempting to send batch of {messages_per_batch} messages. " + f"Batch ID: {current_id // messages_per_batch}" + ) + try: + await client.send_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + partitioning=PARTITION_ID, + messages=messages, + ) + n_sent_batches += 1 + logger.info( + f"Successfully sent batch of {messages_per_batch} messages. " + f"Batch ID: {current_id // messages_per_batch}" + ) + except Exception as error: + logger.error(f"Exception type: {type(error).__name__}, message: {error}") + logger.exception(error) + break + + await asyncio.sleep(interval) + logger.info(f"Sent {n_sent_batches} batches of messages, exiting.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/foreign/python/README.md b/foreign/python/README.md index dbbdbe02ca..43eed6a3bb 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -166,6 +166,31 @@ async def main(): await client.connect() +asyncio.run(main()) +``` + +`IggyClient.http(...)` takes an `HttpConfig` the same way, built from `IggyClient.http()`'s own +config type rather than passed to `IggyClient(...)`. HTTP is a stateless per-request transport, +so there is no `AutoLogin` or reconnection policy to configure: + +```python +import asyncio + +from apache_iggy import HttpConfig, IggyClient + + +async def main(): + client = IggyClient.http( + HttpConfig( + api_url="http://127.0.0.1:3000", + retries=3, + # jwt="...", + ) + ) + await client.connect() + await client.login_user("iggy", "iggy") + + asyncio.run(main()) ``` diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 254b23f41c..6adf0ac839 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -37,6 +37,7 @@ __all__ = [ "GlobalPermissions", "HeaderKey", "HeaderValue", + "HttpConfig", "IggyClient", "IggyConsumer", "IggyExpiry", @@ -837,6 +838,48 @@ class HeaderValue: def value(self) -> builtins.float: ... def __new__(cls, value: builtins.float) -> HeaderValue.Float64: ... +@typing.final +class HttpConfig: + r""" + Configuration for the HTTP transport, accepted by `IggyClient.http(...)`. + + Every field is keyword-only and optional. + """ + @property + def api_url(self) -> builtins.str: ... + @property + def retries(self) -> builtins.int: ... + @property + def has_jwt(self) -> builtins.bool: + r""" + Whether a JWT is configured, without exposing the token itself. + """ + @property + def heartbeat_interval(self) -> datetime.timedelta: ... + def __new__( + cls, + *, + api_url: builtins.str | None = None, + retries: builtins.int | None = None, + jwt: builtins.str | None = None, + heartbeat_interval: datetime.timedelta | None = None, + ) -> HttpConfig: + r""" + Constructs an HTTP configuration. + + Args: + api_url: Base URL of the Iggy HTTP API. Defaults to `http://127.0.0.1:3000`. + retries: Number of retries to perform on transient errors. Defaults to 3. + jwt: JWT token for A2A (Agent-to-Agent) authentication. Defaults to `None`. + heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + + Raises: + ValueError: If `api_url` is not a valid URL, if `retries` is outside the + range of an unsigned 32-bit integer, if a duration is negative, or + if `heartbeat_interval` is zero. + """ + def __repr__(self) -> builtins.str: ... + @typing.final class IggyClient: r""" @@ -867,6 +910,21 @@ class IggyClient: Constructs a new IggyClient from a connection string. Returns an error if the connection string provided is invalid. """ + @classmethod + def http(cls, config: HttpConfig | None = None) -> IggyClient: + r""" + Constructs a new IggyClient configured for the HTTP transport. + + `api_url` is already validated when `config` is built, so this does not + currently fail; the exception is documented for interface consistency + with the other transport constructors. + + Args: + config: HTTP transport configuration. Defaults to `HttpConfig()`. + + Raises: + RuntimeError: If the client cannot be constructed. + """ def ping(self) -> collections.abc.Awaitable[None]: r""" Sends a ping request to the server to check connectivity. diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index f669a87484..b7587cae85 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -31,7 +31,7 @@ use std::collections::BTreeMap; use std::str::FromStr; use std::sync::Arc; -use crate::config::PyClientConfig; +use crate::config::{HttpConfig, PyClientConfig}; use crate::consumer::{ AutoCommit, Consumer as PyConsumer, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer, @@ -144,6 +144,30 @@ impl IggyClient { }) } + /// Constructs a new IggyClient configured for the HTTP transport. + /// + /// `api_url` is already validated when `config` is built, so this does not + /// currently fail; the exception is documented for interface consistency + /// with the other transport constructors. + /// + /// Args: + /// config: HTTP transport configuration. Defaults to `HttpConfig()`. + /// + /// Raises: + /// RuntimeError: If the client cannot be constructed. + #[classmethod] + #[pyo3(signature = (config=None))] + fn http(_cls: &Bound<'_, PyType>, config: Option) -> PyResult { + let config = config + .map(|config| config.client_config()) + .unwrap_or_else(|| Arc::new(HttpClientConfig::default())); + let http_client = HttpClient::create(config) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(RustIggyClient::new(ClientWrapper::Http(http_client))), + }) + } + /// Sends a ping request to the server to check connectivity. /// Raises `RuntimeError` if the connection fails. #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs index ed6e2a14f0..e8d23726fc 100644 --- a/foreign/python/src/config.rs +++ b/foreign/python/src/config.rs @@ -17,6 +17,7 @@ use iggy::prelude::{ AutoLogin as RustAutoLogin, Credentials as RustCredentials, + HttpClientConfig as RustHttpClientConfig, HttpClientConfigBuilder, TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, }; @@ -408,6 +409,110 @@ impl TcpConfig { } } +/// Configuration for the HTTP transport, accepted by `IggyClient.http(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct HttpConfig { + inner: Arc, +} + +impl HttpConfig { + /// The configuration in the shape `HttpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl HttpConfig { + /// Constructs an HTTP configuration. + /// + /// Args: + /// api_url: Base URL of the Iggy HTTP API. Defaults to `http://127.0.0.1:3000`. + /// retries: Number of retries to perform on transient errors. Defaults to 3. + /// jwt: JWT token for A2A (Agent-to-Agent) authentication. Defaults to `None`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If `api_url` is not a valid URL, if `retries` is outside the + /// range of an unsigned 32-bit integer, if a duration is negative, or + /// if `heartbeat_interval` is zero. + #[new] + #[pyo3(signature = (*, api_url=None, retries=None, jwt=None, heartbeat_interval=None))] + fn new( + #[gen_stub(override_type(type_repr = "builtins.str | None"))] api_url: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] retries: Option, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] jwt: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option>, + ) -> PyResult { + // The builder starts from `HttpClientConfig::default()`, and its `build()` + // trims and validates the API URL whether or not one was set here. + let mut builder = HttpClientConfigBuilder::new(); + if let Some(api_url) = api_url { + builder = builder.with_api_url(api_url); + } + if let Some(retries) = retries { + let retries = u32::try_from(retries).map_err(|_| { + PyValueError::new_err(format!("'retries' must be between 0 and {}", u32::MAX)) + })?; + builder = builder.with_retries(retries); + } + if let Some(jwt) = jwt { + builder = builder.with_jwt(jwt); + } + let mut inner = builder + .build() + .map_err(|e| PyValueError::new_err(e.to_string()))?; + if let Some(heartbeat_interval) = heartbeat_interval { + inner.heartbeat_interval = reject_zero( + py_delta_to_iggy_duration(&heartbeat_interval)?, + "heartbeat_interval", + )?; + } + + Ok(Self { + inner: Arc::new(inner), + }) + } + + #[getter] + fn api_url(&self) -> String { + self.inner.api_url.clone() + } + + #[getter] + fn retries(&self) -> u32 { + self.inner.retries + } + + /// Whether a JWT is configured, without exposing the token itself. + #[getter] + fn has_jwt(&self) -> bool { + self.inner.jwt.is_some() + } + + #[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.get()) + } + + fn __repr__(&self) -> String { + let jwt = if self.inner.jwt.is_some() { "..." } else { "None" }; + format!( + "HttpConfig(api_url={:?}, retries={}, jwt={jwt}, heartbeat_interval={})", + self.inner.api_url, + self.inner.retries, + duration_repr(self.inner.heartbeat_interval.get()), + ) + } +} + fn python_bool(value: bool) -> &'static str { if value { "True" } else { "False" } } diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index 5f2e128264..81ff98a88f 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -30,7 +30,7 @@ mod user; mod user_headers; use client::IggyClient; -use config::{AutoLogin, TcpConfig, TcpReconnectionConfig}; +use config::{AutoLogin, HttpConfig, TcpConfig, TcpReconnectionConfig}; use consumer::{ AutoCommit, AutoCommitAfter, AutoCommitWhen, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator, @@ -56,6 +56,7 @@ 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::()?; diff --git a/foreign/python/tests/test_http_config.py b/foreign/python/tests/test_http_config.py new file mode 100644 index 0000000000..b14223d0d8 --- /dev/null +++ b/foreign/python/tests/test_http_config.py @@ -0,0 +1,145 @@ +# 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 HTTP client configuration surface. + +`HttpConfig` mirrors the Rust SDK's `HttpClientConfig` the same way +`TcpConfig`/`QuicConfig` do, 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. HTTP is stateless per-request, so unlike TCP/QUIC there is no +`AutoLogin` or reconnection policy to configure. +""" + +import ast +from datetime import timedelta + +import pytest + +from apache_iggy import HttpConfig, IggyClient + +from .utils import get_http_server_config, wait_for_ping + + +@pytest.mark.unit +class TestHttpConfig: + """Test the transport configuration.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured transport matches the Rust SDK defaults.""" + config = HttpConfig() + + assert config.api_url == "http://127.0.0.1:3000" + assert config.retries == 3 + assert config.has_jwt is False + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + config = HttpConfig( + api_url="http://127.0.0.1:3001", + retries=5, + jwt="a-token", + ) + + assert config.api_url == "http://127.0.0.1:3001" + assert config.retries == 5 + assert config.has_jwt is True + + def test_arguments_are_keyword_only(self): + """Test that the API URL cannot be passed positionally.""" + with pytest.raises(TypeError): + # pyrefly: ignore # bad-argument-count + HttpConfig("http://127.0.0.1:3000") + + def test_repr_hides_the_jwt(self): + """Test that the JWT does not leak through repr but still parses as Python.""" + config = HttpConfig(jwt="a-secret-token") + + printed = repr(config) + + assert "a-secret-token" not in printed + ast.parse(printed) + + def test_repr_shows_every_field_as_python(self): + """Test that repr covers the configured fields and parses as Python.""" + config = HttpConfig(api_url="http://127.0.0.1:3001", retries=5) + + printed = repr(config) + + assert 'api_url="http://127.0.0.1:3001"' in printed + assert "retries=5" in printed + ast.parse(printed) + + @pytest.mark.parametrize( + "invalid_url", + ["", "not-a-url", "http://127.0.0.1:0"], + ) + def test_invalid_api_url_is_rejected(self, invalid_url: str): + """Test that a malformed API URL fails at construction, not at connect.""" + with pytest.raises(ValueError): + HttpConfig(api_url=invalid_url) + + @pytest.mark.parametrize("out_of_range", [-1, 2**32]) + def test_out_of_range_retries_is_rejected(self, out_of_range: int): + """Test that a retry count outside the wire range names the argument. + + The conversion pyo3 does on its own raises OverflowError, which is not a + ValueError and so escapes the handler a caller wraps construction in. + """ + with pytest.raises(ValueError, match="retries"): + HttpConfig(retries=out_of_range) + + def test_negative_heartbeat_interval_is_rejected(self): + """Test that a negative heartbeat interval fails at construction.""" + with pytest.raises(ValueError, match="negative"): + HttpConfig(heartbeat_interval=timedelta(seconds=-3)) + + def test_zero_heartbeat_interval_is_rejected(self): + """Test that a zero heartbeat interval fails at construction. + + Nothing downstream reads zero as "disabled"; it heartbeats in a + continuous loop for as long as the client lives. + """ + with pytest.raises(ValueError, match="zero"): + HttpConfig(heartbeat_interval=timedelta(0)) + + +@pytest.mark.unit +class TestClientConstruction: + """Test what `IggyClient.http(...)` accepts.""" + + def test_accepts_a_config(self): + """Test that a client can be built from a config object.""" + assert IggyClient.http(HttpConfig(api_url="http://127.0.0.1:3000")) is not None + + def test_accepts_nothing(self): + """Test that the default configuration is used when no argument is given.""" + assert IggyClient.http() is not None + + +@pytest.mark.integration +class TestHttpConfigAgainstServer: + """Test that a client built from `HttpConfig` actually connects.""" + + @pytest.mark.asyncio + async def test_client_connects_and_pings(self): + """Test that a client built with a custom config reaches the server.""" + host, port = get_http_server_config() + + client = IggyClient.http(HttpConfig(api_url=f"http://{host}:{port}")) + await client.connect() + await wait_for_ping(client) diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py index b37a53831e..307d1c2d90 100644 --- a/foreign/python/tests/utils.py +++ b/foreign/python/tests/utils.py @@ -33,15 +33,19 @@ MAX_PASSWORD_BYTES = 100 -def get_server_config() -> tuple[str, int]: +def get_transport_config(port_env_var: str, default_port: int) -> tuple[str, int]: """ - Get server configuration from environment variables or defaults. + Get transport-specific server configuration from environment variables or defaults. + + Args: + port_env_var: Name of the environment variable holding the port. + default_port: Port to use if the environment variable is not set. Returns: tuple: (host, port) for the Iggy server """ host = os.environ.get("IGGY_SERVER_HOST", "127.0.0.1") - port = int(os.environ.get("IGGY_SERVER_TCP_PORT", "8090")) + port = int(os.environ.get(port_env_var, str(default_port))) # Convert hostname to IP address for the Rust client if host not in ("127.0.0.1", "localhost"): @@ -58,6 +62,26 @@ def get_server_config() -> tuple[str, int]: return host, port +def get_server_config() -> tuple[str, int]: + """ + Get TCP server configuration from environment variables or defaults. + + Returns: + tuple: (host, port) for the Iggy server + """ + return get_transport_config("IGGY_SERVER_TCP_PORT", 8090) + + +def get_http_server_config() -> tuple[str, int]: + """ + Get HTTP server configuration from environment variables or defaults. + + Returns: + tuple: (host, port) for the Iggy server + """ + return get_transport_config("IGGY_SERVER_HTTP_PORT", 3000) + + def wait_for_server(host: str, port: int, timeout: int = 60, interval: int = 2) -> None: """ Wait for the server to become available. From 10e7fc43573760739da18a1f37f9ed65d2b9c8eb Mon Sep 17 00:00:00 2001 From: saie-ch <132209179+saie-ch@users.noreply.github.com> Date: Sat, 29 Aug 2026 21:56:15 +0530 Subject: [PATCH 2/3] fix(python): rustfmt HttpConfig repr --- foreign/python/src/config.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs index e8d23726fc..f283c0a091 100644 --- a/foreign/python/src/config.rs +++ b/foreign/python/src/config.rs @@ -503,7 +503,11 @@ impl HttpConfig { } fn __repr__(&self) -> String { - let jwt = if self.inner.jwt.is_some() { "..." } else { "None" }; + let jwt = if self.inner.jwt.is_some() { + "..." + } else { + "None" + }; format!( "HttpConfig(api_url={:?}, retries={}, jwt={jwt}, heartbeat_interval={})", self.inner.api_url, From 2eaaadaee840a4a2f1e542f644413d26a7a91148 Mon Sep 17 00:00:00 2001 From: saie-ch <132209179+saie-ch@users.noreply.github.com> Date: Mon, 31 Aug 2026 09:48:35 +0530 Subject: [PATCH 3/3] fix(python): address HTTP PR review feedback Assert heartbeat_interval in config tests, add a real HTTP send/poll round trip, rename TestClientConstruction to avoid colliding with test_client_config.py, and drop the inaccurate "stateless" framing around the missing AutoLogin. --- examples/python/http/consumer.py | 4 +- examples/python/http/producer.py | 4 +- foreign/python/README.md | 3 +- foreign/python/tests/test_http_config.py | 49 +++++++++++++++++++++++- 4 files changed, 53 insertions(+), 7 deletions(-) diff --git a/examples/python/http/consumer.py b/examples/python/http/consumer.py index ef5bd73ba1..1df6574f65 100644 --- a/examples/python/http/consumer.py +++ b/examples/python/http/consumer.py @@ -98,8 +98,8 @@ async def main(): logger.info("Connecting to IggyClient...") await client.connect() logger.info("Connected.") - # HTTP is a stateless transport: log in explicitly rather than relying - # on auto-login, which HttpConfig does not expose. + # Log in explicitly rather than relying on auto-login, which + # HttpConfig does not expose. await client.login_user("iggy", "iggy") await consume_messages(client) except Exception as error: diff --git a/examples/python/http/producer.py b/examples/python/http/producer.py index 7355e76421..78311f7d46 100644 --- a/examples/python/http/producer.py +++ b/examples/python/http/producer.py @@ -91,8 +91,8 @@ async def main(): logger.info("Connecting to IggyClient") await client.connect() logger.info("Connected.") - # HTTP is a stateless transport: log in explicitly rather than relying - # on auto-login, which HttpConfig does not expose. + # Log in explicitly rather than relying on auto-login, which HttpConfig + # does not expose. await client.login_user("iggy", "iggy") await init_system(client) await produce_messages(client) diff --git a/foreign/python/README.md b/foreign/python/README.md index 43eed6a3bb..2353e5729f 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -171,7 +171,8 @@ asyncio.run(main()) `IggyClient.http(...)` takes an `HttpConfig` the same way, built from `IggyClient.http()`'s own config type rather than passed to `IggyClient(...)`. HTTP is a stateless per-request transport, -so there is no `AutoLogin` or reconnection policy to configure: +so there is no reconnection policy to configure. There is also no `AutoLogin`: call +`login_user(...)` explicitly after connecting. ```python import asyncio diff --git a/foreign/python/tests/test_http_config.py b/foreign/python/tests/test_http_config.py index b14223d0d8..7e4315b9d9 100644 --- a/foreign/python/tests/test_http_config.py +++ b/foreign/python/tests/test_http_config.py @@ -30,7 +30,8 @@ import pytest -from apache_iggy import HttpConfig, IggyClient +from apache_iggy import Consumer, HttpConfig, IggyClient, PollingStrategy +from apache_iggy import SendMessage as Message from .utils import get_http_server_config, wait_for_ping @@ -46,6 +47,7 @@ def test_defaults_match_the_rust_sdk(self): assert config.api_url == "http://127.0.0.1:3000" assert config.retries == 3 assert config.has_jwt is False + assert config.heartbeat_interval == timedelta(seconds=5) def test_every_field_round_trips(self): """Test that each configured field is readable back unchanged.""" @@ -53,11 +55,13 @@ def test_every_field_round_trips(self): api_url="http://127.0.0.1:3001", retries=5, jwt="a-token", + heartbeat_interval=timedelta(seconds=15), ) assert config.api_url == "http://127.0.0.1:3001" assert config.retries == 5 assert config.has_jwt is True + assert config.heartbeat_interval == timedelta(seconds=15) def test_arguments_are_keyword_only(self): """Test that the API URL cannot be passed positionally.""" @@ -119,7 +123,7 @@ def test_zero_heartbeat_interval_is_rejected(self): @pytest.mark.unit -class TestClientConstruction: +class TestHttpClientConstruction: """Test what `IggyClient.http(...)` accepts.""" def test_accepts_a_config(self): @@ -143,3 +147,44 @@ async def test_client_connects_and_pings(self): client = IggyClient.http(HttpConfig(api_url=f"http://{host}:{port}")) await client.connect() await wait_for_ping(client) + + @pytest.mark.asyncio + async def test_client_sends_and_polls_a_message(self, unique_name): + """Test a full round trip: login, create stream/topic, send, poll. + + HTTP is a stateless per-request transport, so this is the part + `test_client_connects_and_pings` above does not cover: that a client + built from `HttpConfig` can actually carry a real workload. + """ + host, port = get_http_server_config() + stream_name = unique_name() + topic_name = unique_name() + payload = f"payload-{unique_name()}" + + client = IggyClient.http(HttpConfig(api_url=f"http://{host}:{port}")) + await client.connect() + await wait_for_ping(client) + await client.login_user("iggy", "iggy") + + await client.create_stream(stream_name) + await client.create_topic( + stream=stream_name, name=topic_name, partitions_count=1 + ) + await client.send_messages( + stream=stream_name, + topic=topic_name, + partitioning=0, + messages=[Message(payload)], + ) + + polled_messages = await client.poll_messages( + stream=stream_name, + topic=topic_name, + consumer=Consumer.Single("http-round-trip"), + partition_id=0, + polling_strategy=PollingStrategy.First(), + count=1, + auto_commit=True, + ) + + assert [message.payload().decode() for message in polled_messages] == [payload]