feat(python): add user header and origin timestamp support

core/sdk/src/prelude.rs

@@ -51,9 +51,9 @@ pub use iggy_common::{
     Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, CacheMetrics, CacheMetricsKey, ClientError,
     ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus,
     CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember,
-    ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, HeaderValue,
-    HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo,
-    IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage,
+    ConsumerKind, EncryptorKind, GlobalPermissions, HeaderField, HeaderKey, HeaderKind,
+    HeaderValue, HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier,
+    IdentityInfo, IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage,
     IggyMessageHeader, IggyMessageHeaderView, IggyMessageView, IggyMessageViewIterator,
     IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning, Permissions,
     PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind, PollingStrategy,

examples/python/README.md

@@ -71,6 +71,29 @@ python basic/consumer.py
 
 Demonstrates fundamental client connection, authentication, batch message sending, and polling with support for TCP/QUIC/HTTP protocols.
 
+### Message Headers
+
+Shows how to attach and read Python SDK user headers with `str`, `bytes`, `bool`, `int`, and `float` values. Two variants share their logic through `message-headers/common.py`:
+
+- `plain-headers/` uses the convenient `dict[str, str | bytes | bool | int | float]` form; the SDK infers a wire type for each value.
+- `typed-headers/` uses explicit `HeaderKey`/`HeaderValue` for full control over the wire type.
+
+Both producers store typed headers on the wire. The plain consumer converts them to Python scalars, while the typed consumer preserves and inspects the explicit header kinds.
+
+```bash
+# Using uv
+uv run message-headers/plain-headers/producer.py
+uv run message-headers/plain-headers/consumer.py
+uv run message-headers/typed-headers/producer.py
+uv run message-headers/typed-headers/consumer.py
+
+# Without using uv
+python message-headers/plain-headers/producer.py
+python message-headers/plain-headers/consumer.py
+python message-headers/typed-headers/producer.py
+python message-headers/typed-headers/consumer.py
+```
+
 ## TLS Examples
 
 To test with a TLS-enabled server, start the server with TLS configured (see main README), then run:

examples/python/basic/producer.py

@@ -60,35 +60,26 @@ async def main():
 
 
 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)
+    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,
+            replication_factor=1,
+        )
+        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):
@@ -127,6 +118,7 @@ async def produce_messages(client: IggyClient):
         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.")

examples/python/getting-started/producer.py

@@ -126,35 +126,26 @@ async def main():
 
 
 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)
+    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,
+            replication_factor=1,
+        )
+        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):
@@ -193,6 +184,7 @@ async def produce_messages(client: IggyClient):
         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.")