From 261062db17e4261af5cf991e2551a17b6c6e7496 Mon Sep 17 00:00:00 2001 From: saie-ch <132209179+saie-ch@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:14:59 +0530 Subject: [PATCH] feat(python): add QUIC, HTTP, and WebSocket transport constructors Closes #2835. Adds explicit IggyClient.tcp/quic/http/websocket() constructors mirroring the Rust builder, fixes a QUIC connection-string bug (endpoint creation needs an active Tokio runtime), and adds tests/examples/CI/docs for all three previously-untested transports. --- .../python-maturin/pre-merge/action.yml | 3 + .github/workflows/coverage-baseline.yml | 3 + examples/python/README.md | 31 +++ examples/python/http/consumer.py | 106 ++++++++++ examples/python/http/producer.py | 136 +++++++++++++ examples/python/quic/consumer.py | 106 ++++++++++ examples/python/quic/producer.py | 136 +++++++++++++ examples/python/websocket/consumer.py | 106 ++++++++++ examples/python/websocket/producer.py | 136 +++++++++++++ foreign/python/README.md | 27 +++ foreign/python/apache_iggy.pyi | 92 +++++++++ foreign/python/docker-compose.test.yml | 2 + foreign/python/src/client.rs | 182 ++++++++++++++++++ foreign/python/tests/test_connectivity.py | 35 +++- .../python/tests/test_transport_operations.py | 133 +++++++++++++ foreign/python/tests/utils.py | 50 ++++- 16 files changed, 1279 insertions(+), 5 deletions(-) create mode 100644 examples/python/http/consumer.py create mode 100644 examples/python/http/producer.py create mode 100644 examples/python/quic/consumer.py create mode 100644 examples/python/quic/producer.py create mode 100644 examples/python/websocket/consumer.py create mode 100644 examples/python/websocket/producer.py create mode 100644 foreign/python/tests/test_transport_operations.py diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml index 81d3bd50b9..8bbf0becde 100644 --- a/.github/actions/python-maturin/pre-merge/action.yml +++ b/.github/actions/python-maturin/pre-merge/action.yml @@ -135,6 +135,9 @@ 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_QUIC_PORT=8080 \ + IGGY_SERVER_WS_PORT=8092 \ uv run --no-sync pytest tests/ -v \ --junitxml=../../reports/python-junit.xml \ --tb=short \ diff --git a/.github/workflows/coverage-baseline.yml b/.github/workflows/coverage-baseline.yml index 1f580db3d6..ea531f8279 100644 --- a/.github/workflows/coverage-baseline.yml +++ b/.github/workflows/coverage-baseline.yml @@ -273,6 +273,9 @@ jobs: cd foreign/python IGGY_SERVER_HOST=127.0.0.1 \ IGGY_SERVER_TCP_PORT=8090 \ + IGGY_SERVER_HTTP_PORT=3000 \ + IGGY_SERVER_QUIC_PORT=8080 \ + IGGY_SERVER_WS_PORT=8092 \ uv run --no-sync pytest tests/ -v \ --junitxml=../../reports/python-junit.xml \ --tb=short \ diff --git a/examples/python/README.md b/examples/python/README.md index 9bf943b75c..336444935c 100644 --- a/examples/python/README.md +++ b/examples/python/README.md @@ -71,6 +71,37 @@ python basic/consumer.py Demonstrates fundamental client connection, authentication, batch message sending, and polling with support for TCP/QUIC/HTTP protocols. +## Transport Protocol Examples + +Each of the non-TCP transports has its own example pair, using the explicit + +`IggyClient.quic()`/`IggyClient.http()`/`IggyClient.websocket()` constructors (`IggyClient(...)` + +already covers TCP). These assume a server started with defaults, which enables all four + +transports (`cargo run --bin iggy-server`, or the `docker run` command above). + +### QUIC + +```bash +uv run quic/producer.py +uv run quic/consumer.py +``` + +### HTTP + +```bash +uv run http/producer.py +uv run http/consumer.py +``` + +### WebSocket + +```bash +uv run websocket/producer.py +uv run websocket/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..c120f1e0dc --- /dev/null +++ b/examples/python/http/consumer.py @@ -0,0 +1,106 @@ +# 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 +from typing import NamedTuple + +from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage +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(NamedTuple): + api_url: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "api_url", + help="Iggy HTTP API URL", + default="http://127.0.0.1:3000", + nargs="?", + type=str, + ) + return ArgNamespace(**vars(parser.parse_args())) + + +async def main(): + args: ArgNamespace = parse_args() + client = IggyClient.http(api_url=args.api_url) + logger.info("Connecting to Iggy over HTTP") + await client.connect() + logger.info("Connected") + await client.login_user("iggy", "iggy") + logger.info("Authenticated") + await consume_messages(client) + + +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, + 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..f50a6a4b0e --- /dev/null +++ b/examples/python/http/producer.py @@ -0,0 +1,136 @@ +# 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 +from typing import NamedTuple + +from apache_iggy import 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(NamedTuple): + api_url: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "api_url", + help="Iggy HTTP API URL", + default="http://127.0.0.1:3000", + nargs="?", + type=str, + ) + return ArgNamespace(**vars(parser.parse_args())) + + +async def main(): + args: ArgNamespace = parse_args() + client = IggyClient.http(api_url=args.api_url) + logger.info("Connecting to Iggy over HTTP") + await client.connect() + logger.info("Connected") + await client.login_user("iggy", "iggy") + logger.info("Authenticated") + await init_system(client) + await produce_messages(client) + + +async def init_system(client: IggyClient): + try: + 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}") + + except Exception as error: + logger.error(f"Error creating stream: {error}") + logger.exception(error) + + try: + 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, + replication_factor=1, + ) + logger.info("Topic was created successfully.") + else: + logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + except Exception as error: + logger.error(f"Error creating topic {error}") + logger.exception(error) + + +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) + + await asyncio.sleep(interval) + logger.info(f"Sent {n_sent_batches} batches of messages, exiting.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/python/quic/consumer.py b/examples/python/quic/consumer.py new file mode 100644 index 0000000000..90c009c254 --- /dev/null +++ b/examples/python/quic/consumer.py @@ -0,0 +1,106 @@ +# 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 +from typing import NamedTuple + +from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage +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(NamedTuple): + server_address: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "server_address", + help="Iggy QUIC server address (host:port)", + default="127.0.0.1:8080", + nargs="?", + type=str, + ) + return ArgNamespace(**vars(parser.parse_args())) + + +async def main(): + args: ArgNamespace = parse_args() + client = IggyClient.quic(server_address=args.server_address) + logger.info("Connecting to Iggy over QUIC") + await client.connect() + logger.info("Connected") + await client.login_user("iggy", "iggy") + logger.info("Authenticated") + await consume_messages(client) + + +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, + 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/quic/producer.py b/examples/python/quic/producer.py new file mode 100644 index 0000000000..e113d01874 --- /dev/null +++ b/examples/python/quic/producer.py @@ -0,0 +1,136 @@ +# 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 +from typing import NamedTuple + +from apache_iggy import 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(NamedTuple): + server_address: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "server_address", + help="Iggy QUIC server address (host:port)", + default="127.0.0.1:8080", + nargs="?", + type=str, + ) + return ArgNamespace(**vars(parser.parse_args())) + + +async def main(): + args: ArgNamespace = parse_args() + client = IggyClient.quic(server_address=args.server_address) + logger.info("Connecting to Iggy over QUIC") + await client.connect() + logger.info("Connected") + await client.login_user("iggy", "iggy") + logger.info("Authenticated") + await init_system(client) + await produce_messages(client) + + +async def init_system(client: IggyClient): + try: + 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}") + + except Exception as error: + logger.error(f"Error creating stream: {error}") + logger.exception(error) + + try: + 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, + replication_factor=1, + ) + logger.info("Topic was created successfully.") + else: + logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + except Exception as error: + logger.error(f"Error creating topic {error}") + logger.exception(error) + + +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) + + await asyncio.sleep(interval) + logger.info(f"Sent {n_sent_batches} batches of messages, exiting.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/python/websocket/consumer.py b/examples/python/websocket/consumer.py new file mode 100644 index 0000000000..b883fc8f40 --- /dev/null +++ b/examples/python/websocket/consumer.py @@ -0,0 +1,106 @@ +# 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 +from typing import NamedTuple + +from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage +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(NamedTuple): + server_address: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "server_address", + help="Iggy WebSocket server address (host:port)", + default="127.0.0.1:8092", + nargs="?", + type=str, + ) + return ArgNamespace(**vars(parser.parse_args())) + + +async def main(): + args: ArgNamespace = parse_args() + client = IggyClient.websocket(server_address=args.server_address) + logger.info("Connecting to Iggy over WebSocket") + await client.connect() + logger.info("Connected") + await client.login_user("iggy", "iggy") + logger.info("Authenticated") + await consume_messages(client) + + +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, + 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/websocket/producer.py b/examples/python/websocket/producer.py new file mode 100644 index 0000000000..290b7dfa05 --- /dev/null +++ b/examples/python/websocket/producer.py @@ -0,0 +1,136 @@ +# 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 +from typing import NamedTuple + +from apache_iggy import 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(NamedTuple): + server_address: str + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "server_address", + help="Iggy WebSocket server address (host:port)", + default="127.0.0.1:8092", + nargs="?", + type=str, + ) + return ArgNamespace(**vars(parser.parse_args())) + + +async def main(): + args: ArgNamespace = parse_args() + client = IggyClient.websocket(server_address=args.server_address) + logger.info("Connecting to Iggy over WebSocket") + await client.connect() + logger.info("Connected") + await client.login_user("iggy", "iggy") + logger.info("Authenticated") + await init_system(client) + await produce_messages(client) + + +async def init_system(client: IggyClient): + try: + 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}") + + except Exception as error: + logger.error(f"Error creating stream: {error}") + logger.exception(error) + + try: + 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, + replication_factor=1, + ) + logger.info("Topic was created successfully.") + else: + logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + except Exception as error: + logger.error(f"Error creating topic {error}") + logger.exception(error) + + +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) + + 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 f01754f6dc..3067f5f5f3 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -58,6 +58,33 @@ maturin develop pytest tests/ -v # Run tests (requires iggy-server running) ``` +## Connecting + +`IggyClient` supports all four transport protocols. Use the matching constructor, or build a +client from a connection string (`iggy+://user:pass@host:port`, e.g. `iggy+quic://...`): + +```python +from apache_iggy import IggyClient + +# TCP (also the `IggyClient(server_address)` shorthand). Default: 127.0.0.1:8090 +client = IggyClient.tcp(server_address="127.0.0.1:8090") + +# QUIC. Default: 127.0.0.1:8080 +client = IggyClient.quic(server_address="127.0.0.1:8080") + +# HTTP. Default: http://127.0.0.1:3000 +client = IggyClient.http(api_url="http://127.0.0.1:3000") + +# WebSocket. Default: 127.0.0.1:8092 +client = IggyClient.websocket(server_address="127.0.0.1:8092") + +# Or from a connection string, which also supports TLS and other options: +client = IggyClient.from_connection_string("iggy+tcp://iggy:iggy@127.0.0.1:8090") +``` + +`tcp()` and `websocket()` also accept `tls_enabled`, `tls_domain`, `tls_ca_file`, and +`tls_validate_certificate` keyword arguments. + ## Examples Refer to the [examples/python/](https://github.com/apache/iggy/tree/master/examples/python) directory for usage examples. diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 5b980c2e0d..68388218e1 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -326,6 +326,98 @@ class IggyClient: Constructs a new IggyClient from a connection string. Returns an error if the connection string provided is invalid. """ + @classmethod + def tcp( + cls, + server_address: builtins.str | 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, + no_delay: builtins.bool = False, + ) -> IggyClient: + r""" + Constructs a new IggyClient configured for the TCP transport. + + Args: + server_address: TCP server address as `host:port`. Defaults to `127.0.0.1:8090`. + tls_enabled: Whether to use TLS when connecting to the server. Defaults to `False`. + tls_domain: Domain to use for TLS when connecting to the server. + tls_ca_file: Path to the CA certificate file for TLS. + tls_validate_certificate: Whether to validate the TLS certificate. Defaults to `True`. + no_delay: Whether to disable Nagle's algorithm on the TCP socket. Defaults to `False`. + + Returns: + A new `IggyClient` configured for TCP. + + Raises: + PyRuntimeError: If the client configuration is invalid. + """ + @classmethod + def quic( + cls, + server_address: builtins.str | None = None, + server_name: builtins.str | None = None, + ) -> IggyClient: + r""" + Constructs a new IggyClient configured for the QUIC transport. + + Args: + server_address: QUIC server address as `host:port`. Defaults to `127.0.0.1:8080`. + server_name: Server name used for the QUIC/TLS handshake. Defaults to `localhost`. + + Returns: + A new `IggyClient` configured for QUIC. + + Raises: + PyRuntimeError: If the client configuration is invalid. + """ + @classmethod + def http( + cls, + api_url: builtins.str | None = None, + retries: builtins.int | None = None, + jwt: builtins.str | None = None, + ) -> IggyClient: + r""" + Constructs a new IggyClient configured for the HTTP transport. + + 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. + + Returns: + A new `IggyClient` configured for HTTP. + + Raises: + PyRuntimeError: If the client configuration is invalid. + """ + @classmethod + def websocket( + cls, + server_address: builtins.str | 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 = False, + ) -> IggyClient: + r""" + Constructs a new IggyClient configured for the WebSocket transport. + + Args: + server_address: WebSocket server address as `host:port`. Defaults to `127.0.0.1:8092`. + tls_enabled: Whether to use TLS when connecting to the server. Defaults to `False`. + tls_domain: Domain to use for TLS when connecting to the server. + tls_ca_file: Path to the CA certificate file for TLS. + tls_validate_certificate: Whether to validate the TLS certificate. Defaults to `False`. + + Returns: + A new `IggyClient` configured for WebSocket. + + Raises: + PyRuntimeError: If the client configuration is invalid. + """ def ping(self) -> collections.abc.Awaitable[None]: r""" Sends a ping request to the server to check connectivity. diff --git a/foreign/python/docker-compose.test.yml b/foreign/python/docker-compose.test.yml index 92ffec6daf..3c09d3a1eb 100644 --- a/foreign/python/docker-compose.test.yml +++ b/foreign/python/docker-compose.test.yml @@ -31,6 +31,7 @@ services: - "3000:3000" - "8080:8080" - "8090:8090" + - "8092:8092" healthcheck: test: [ "CMD", "curl", "-f", "http://localhost:3000/stats" ] interval: 5s @@ -55,6 +56,7 @@ services: - IGGY_SERVER_TCP_PORT=8090 - IGGY_SERVER_HTTP_PORT=3000 - IGGY_SERVER_QUIC_PORT=8080 + - IGGY_SERVER_WS_PORT=8092 - PYTHONPATH=/workspace/foreign/python - PYTEST_ARGS=-v --tb=short volumes: diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index b860f19b5d..decb3eb16f 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -83,6 +83,10 @@ impl IggyClient { _cls: &Bound<'_, PyType>, connection_string: String, ) -> PyResult { + // The QUIC transport builds its endpoint eagerly and needs a Tokio runtime context to do + // so (see `quic()` below for details); entering it here is a no-op for the other + // transports since the protocol isn't known until the connection string is parsed. + let _guard = pyo3_async_runtimes::tokio::get_runtime().enter(); let client = RustIggyClient::from_connection_string(&connection_string) .map_err(|e| PyErr::new::(e.to_string()))?; Ok(Self { @@ -90,6 +94,184 @@ impl IggyClient { }) } + /// Constructs a new IggyClient configured for the TCP transport. + /// + /// Args: + /// server_address: TCP server address as `host:port`. Defaults to `127.0.0.1:8090`. + /// tls_enabled: Whether to use TLS when connecting to the server. Defaults to `False`. + /// tls_domain: Domain to use for TLS when connecting to the server. + /// tls_ca_file: Path to the CA certificate file for TLS. + /// tls_validate_certificate: Whether to validate the TLS certificate. Defaults to `True`. + /// no_delay: Whether to disable Nagle's algorithm on the TCP socket. Defaults to `False`. + /// + /// Returns: + /// A new `IggyClient` configured for TCP. + /// + /// Raises: + /// PyRuntimeError: If the client configuration is invalid. + #[classmethod] + #[pyo3(signature = (server_address=None, tls_enabled=false, tls_domain=None, tls_ca_file=None, tls_validate_certificate=true, no_delay=false))] + #[allow(clippy::too_many_arguments)] + fn tcp( + _cls: &Bound<'_, PyType>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option< + String, + >, + 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, + no_delay: bool, + ) -> PyResult { + let mut builder = IggyClientBuilder::new() + .with_tcp() + .with_tls_enabled(tls_enabled) + .with_tls_validate_certificate(tls_validate_certificate); + if let Some(server_address) = server_address { + builder = builder.with_server_address(server_address); + } + if let Some(tls_domain) = tls_domain { + builder = builder.with_tls_domain(tls_domain); + } + if let Some(tls_ca_file) = tls_ca_file { + builder = builder.with_tls_ca_file(tls_ca_file); + } + if no_delay { + builder = builder.with_no_delay(); + } + let client = builder + .build() + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(client), + }) + } + + /// Constructs a new IggyClient configured for the QUIC transport. + /// + /// Args: + /// server_address: QUIC server address as `host:port`. Defaults to `127.0.0.1:8080`. + /// server_name: Server name used for the QUIC/TLS handshake. Defaults to `localhost`. + /// + /// Returns: + /// A new `IggyClient` configured for QUIC. + /// + /// Raises: + /// PyRuntimeError: If the client configuration is invalid. + #[classmethod] + #[pyo3(signature = (server_address=None, server_name=None))] + fn quic( + _cls: &Bound<'_, PyType>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option< + String, + >, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_name: Option, + ) -> PyResult { + let mut builder = IggyClientBuilder::new().with_quic(); + if let Some(server_address) = server_address { + builder = builder.with_server_address(server_address); + } + if let Some(server_name) = server_name { + builder = builder.with_server_name(server_name); + } + // `quinn::Endpoint::client` (invoked eagerly by `.build()`) looks up the current Tokio + // runtime via `Handle::try_current()` and fails with `CannotCreateEndpoint` if none is + // active. This method runs synchronously from Python without one, so enter the runtime + // pyo3-async-runtimes uses for our own async methods before building the endpoint. + let _guard = pyo3_async_runtimes::tokio::get_runtime().enter(); + let client = builder + .build() + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(client), + }) + } + + /// Constructs a new IggyClient configured for the HTTP transport. + /// + /// 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. + /// + /// Returns: + /// A new `IggyClient` configured for HTTP. + /// + /// Raises: + /// PyRuntimeError: If the client configuration is invalid. + #[classmethod] + #[pyo3(signature = (api_url=None, retries=None, jwt=None))] + fn http( + _cls: &Bound<'_, PyType>, + #[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, + ) -> PyResult { + let mut builder = IggyClientBuilder::new().with_http(); + if let Some(api_url) = api_url { + builder = builder.with_api_url(api_url); + } + if let Some(retries) = retries { + builder = builder.with_retries(retries); + } + if let Some(jwt) = jwt { + builder = builder.with_jwt(jwt); + } + let client = builder + .build() + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(client), + }) + } + + /// Constructs a new IggyClient configured for the WebSocket transport. + /// + /// Args: + /// server_address: WebSocket server address as `host:port`. Defaults to `127.0.0.1:8092`. + /// tls_enabled: Whether to use TLS when connecting to the server. Defaults to `False`. + /// tls_domain: Domain to use for TLS when connecting to the server. + /// tls_ca_file: Path to the CA certificate file for TLS. + /// tls_validate_certificate: Whether to validate the TLS certificate. Defaults to `False`. + /// + /// Returns: + /// A new `IggyClient` configured for WebSocket. + /// + /// Raises: + /// PyRuntimeError: If the client configuration is invalid. + #[classmethod] + #[pyo3(signature = (server_address=None, tls_enabled=false, tls_domain=None, tls_ca_file=None, tls_validate_certificate=false))] + fn websocket( + _cls: &Bound<'_, PyType>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option< + String, + >, + 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, + ) -> PyResult { + let mut builder = IggyClientBuilder::new() + .with_websocket() + .with_tls_enabled(tls_enabled) + .with_tls_validate_certificate(tls_validate_certificate); + if let Some(server_address) = server_address { + builder = builder.with_server_address(server_address); + } + if let Some(tls_domain) = tls_domain { + builder = builder.with_tls_domain(tls_domain); + } + if let Some(tls_ca_file) = tls_ca_file { + builder = builder.with_tls_ca_file(tls_ca_file); + } + let client = builder + .build() + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(client), + }) + } + /// Sends a ping request to the server to check connectivity. /// Returns `Ok(())` if the server responds successfully, or a `PyRuntimeError` /// if the connection fails. diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 69516d74c1..feba881be4 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -19,7 +19,14 @@ from apache_iggy import IggyClient -from .utils import get_server_config, wait_for_ping, wait_for_server +from .utils import ( + get_http_server_config, + get_quic_server_config, + get_server_config, + get_websocket_server_config, + wait_for_ping, + wait_for_server, +) class TestConnectivity: @@ -40,6 +47,7 @@ async def test_client_not_none(self, iggy_client: IggyClient): "iggy+http://iggy:iggy@127.0.0.1:3000?heartbeat_interval=5s&retries=3", "iggy+ws://iggy:iggy@127.0.0.1:8092", "iggy+ws://iggy:iggy@127.0.0.1:8092?heartbeat_interval=5s&reconnection_retries=3&reconnection_interval=1s&reestablish_after=5s&read_buffer_size=4096&write_buffer_size=4096&max_write_buffer_size=8192&max_message_size=16384&max_frame_size=16384&accept_unmasked_frames=false&tls_domain=localhost&tls_ca_file=unused.pem&tls_validate_certificate=false&tls=false", + "iggy+quic://iggy:iggy@127.0.0.1:8080", ], ) @pytest.mark.asyncio @@ -50,6 +58,30 @@ async def test_valid_connection_string(self, connection_string: str): await client.connect() await wait_for_ping(client, timeout=5, interval=1) + @pytest.mark.parametrize("transport", ["tcp", "quic", "http", "websocket"]) + @pytest.mark.asyncio + async def test_explicit_transport_constructor_connects(self, transport: str): + """Test each per-transport constructor connects and responds to ping.""" + if transport == "tcp": + host, port = get_server_config() + wait_for_server(host, port) + client = IggyClient.tcp(server_address=f"{host}:{port}") + elif transport == "quic": + # QUIC is UDP, so `wait_for_server`'s TCP connect check doesn't apply. + host, port = get_quic_server_config() + client = IggyClient.quic(server_address=f"{host}:{port}") + elif transport == "http": + host, port = get_http_server_config() + wait_for_server(host, port) + client = IggyClient.http(api_url=f"http://{host}:{port}") + else: + host, port = get_websocket_server_config() + wait_for_server(host, port) + client = IggyClient.websocket(server_address=f"{host}:{port}") + + await client.connect() + await wait_for_ping(client, timeout=5, interval=1) + @pytest.mark.parametrize( ("invalid_value", "expected_error"), [ @@ -77,7 +109,6 @@ async def test_valid_connection_string(self, connection_string: str): "iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value", "Invalid connection string", ), - ("iggy+quic://iggy:iggy@127.0.0.1:8080", "Cannot create endpoint"), ], ) def test_invalid_connection_string(self, invalid_value: str, expected_error: str): diff --git a/foreign/python/tests/test_transport_operations.py b/foreign/python/tests/test_transport_operations.py new file mode 100644 index 0000000000..7605423945 --- /dev/null +++ b/foreign/python/tests/test_transport_operations.py @@ -0,0 +1,133 @@ +# 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. + +""" +Integration tests for stream creation and message produce/consume over the QUIC, HTTP, +and WebSocket transports, parametrized like the other per-feature test files. + +TCP already has equivalent coverage via the session-scoped `iggy_client` fixture used +throughout test_stream.py and test_message_operations.py; basic connect+ping coverage +for all four transports (including the explicit constructors added alongside this file) +lives in test_connectivity.py. +""" + +import pytest + +from apache_iggy import IggyClient, PollingStrategy +from apache_iggy import SendMessage as Message + +from .utils import ( + get_http_server_config, + get_quic_server_config, + get_websocket_server_config, + wait_for_ping, + wait_for_server, +) + + +async def _quic_client() -> IggyClient: + host, port = get_quic_server_config() + client = IggyClient.quic(server_address=f"{host}:{port}") + await client.connect() + await wait_for_ping(client) + await client.login_user("iggy", "iggy") + return client + + +async def _http_client() -> IggyClient: + host, port = get_http_server_config() + wait_for_server(host, port) + client = IggyClient.http(api_url=f"http://{host}:{port}") + await client.connect() + await wait_for_ping(client) + await client.login_user("iggy", "iggy") + return client + + +async def _websocket_client() -> IggyClient: + host, port = get_websocket_server_config() + wait_for_server(host, port) + client = IggyClient.websocket(server_address=f"{host}:{port}") + await client.connect() + await wait_for_ping(client) + await client.login_user("iggy", "iggy") + return client + + +TRANSPORT_CLIENT_FACTORIES = { + "quic": _quic_client, + "http": _http_client, + "websocket": _websocket_client, +} + + +@pytest.fixture +async def transport_client(request) -> IggyClient: + """Build an authenticated client for the transport named by the parametrize id.""" + return await TRANSPORT_CLIENT_FACTORIES[request.param]() + + +@pytest.mark.integration +class TestTransportOperations: + """Test stream creation and message produce/consume over QUIC, HTTP, WebSocket.""" + + @pytest.mark.parametrize( + "transport_client", ["quic", "http", "websocket"], indirect=True + ) + @pytest.mark.asyncio + async def test_create_stream(self, transport_client: IggyClient, unique_name): + """Test creating and getting a stream over the given transport.""" + stream_name = unique_name() + await transport_client.create_stream(stream_name) + stream = await transport_client.get_stream(stream_name) + assert stream is not None + + @pytest.mark.parametrize( + "transport_client", ["quic", "http", "websocket"], indirect=True + ) + @pytest.mark.asyncio + async def test_produce_and_consume(self, transport_client: IggyClient, unique_name): + """Test producing and consuming messages over the given transport.""" + stream_name = unique_name() + topic_name = unique_name() + partition_id = 0 + + await transport_client.create_stream(stream_name) + await transport_client.create_topic(stream_name, topic_name, partitions_count=1) + + test_messages = [f"message-{i}" for i in range(3)] + messages = [Message(msg) for msg in test_messages] + await transport_client.send_messages( + stream=stream_name, + topic=topic_name, + partitioning=partition_id, + messages=messages, + ) + + polled = await transport_client.poll_messages( + stream=stream_name, + topic=topic_name, + partition_id=partition_id, + polling_strategy=PollingStrategy.First(), + count=10, + auto_commit=True, + ) + assert len(polled) >= len(test_messages) + + for i, expected_msg in enumerate(test_messages): + if i < len(polled): + assert polled[i].payload().decode("utf-8") == expected_msg diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py index bbddc29de1..dd14f91cfd 100644 --- a/foreign/python/tests/utils.py +++ b/foreign/python/tests/utils.py @@ -27,15 +27,19 @@ from apache_iggy import IggyClient -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"): @@ -52,6 +56,46 @@ 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_quic_server_config() -> tuple[str, int]: + """ + Get QUIC server configuration from environment variables or defaults. + + Returns: + tuple: (host, port) for the Iggy server + """ + return get_transport_config("IGGY_SERVER_QUIC_PORT", 8080) + + +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 get_websocket_server_config() -> tuple[str, int]: + """ + Get WebSocket server configuration from environment variables or defaults. + + Returns: + tuple: (host, port) for the Iggy server + """ + return get_transport_config("IGGY_SERVER_WS_PORT", 8092) + + def wait_for_server(host: str, port: int, timeout: int = 60, interval: int = 2) -> None: """ Wait for the server to become available.