feat(python): add QUIC, HTTP, and WebSocket transport support

.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 \

.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 \

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:

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())

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())