This is an automated email from the ASF dual-hosted git repository.
slbotbm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new a684e550f feat(python): add user header and origin timestamp support
(#3613)
a684e550f is described below
commit a684e550f6736b936dadda13327961de1f82d472
Author: Gunther Xing <[email protected]>
AuthorDate: Tue Aug 4 06:28:59 2026 +0800
feat(python): add user header and origin timestamp support (#3613)
Add user headers to SendMessage and ReceiveMessage, expose origin
timestamp, and fix Docker test infrastructure.
## Which issue does this PR address?
<!--
We generally require a GitHub issue for all bug fixes and enhancements.
Link it with GitHub syntax, keep the line that applies and delete the
other:
- `Closes #123` auto-closes the issue when this PR merges (full fix).
- `Relates to #123` links without closing (partial or related work).
-->
Closes #3601 #3612
## Rationale
<!--
Why is this change needed? If the issue explains it well, a one-liner is
fine.
-->
The Python SDK could send and receive message payloads, but it did not
expose the typed user headers already carried by the underlying Rust
`IggyMessage`. This left Python behind the Rust, Node, Go, Java, and C#
SDKs for message metadata.
## What changed?
Python messages can attach user headers and read them back from received
messages. `SendMessage` accepts `user_headers` and an optional custom
`id`, while `ReceiveMessage` exposes `user_headers()` and
`origin_timestamp()`.
According to the discussion below, unlike the original proposal in #3601
(which exposed only a plain `dict[str, str | bytes | bool | int |
float]`), the binding now supports the **full Rust header type surface**
through two exposed classes, plus a plain-scalar convenience layer:
- **`HeaderKey` / `HeaderValue`** — typed complex enums covering every
`HeaderKind`: `Raw`, `String`, `Bool`, `Int8/16/32/64/128`,
`UnsignedInt8/16/32/64/128`, `Float32`, `Float64`.
- **`UserHeaders`** — the mapping returned by
`ReceiveMessage.user_headers()`. It is a real `dict` subclass
(`dict[HeaderKey, HeaderValue]`), so all mapping operations work, and it
adds a chainable `to_scalar_dict()` for the convenient scalar form:
`message.user_headers().to_scalar_dict()`.
### Sending (`plain → Rust`)
Each key/value pair is converted **independently**, so typed and plain
forms can be freely **mixed** in one dict (e.g. `{HeaderKey.String("a"):
7, "b": HeaderValue.Bool(True), "c": "plain"}`):
- Plain scalars map by best-effort, **lossless** conversion for both
keys and values (keys are no longer restricted to `str`):
- `str → String`, `bytes → Raw`, `bool → Bool`
- `int →` the **smallest** header kind that holds it exactly:
non-negative → `Uint8/16/32/64/128`, negative → `Int8/16/32/64/128`;
values beyond the 128-bit range raise `ValueError`.
- `float → Float32` when the value is exactly representable as f32,
otherwise `Float64` (always lossless).
- Explicit typed values are **range-checked**: integer variants are
enforced by the binding at construction (`OverflowError` for
out-of-range, e.g. `HeaderValue.Int8(300)`), and an explicit `Float32`
whose value overflows f32 (→ `inf`) is rejected with `ValueError`.
### Receiving (`Rust → plain`)
Every `HeaderKind` maps **losslessly** onto a Python scalar (128-bit
ints fit Python's arbitrary-precision `int`; `f32 → f64` is exact), so:
- `user_headers()` returns typed `UserHeaders` (`dict[HeaderKey,
HeaderValue]`), or `None` when the message carries no user headers.
- `UserHeaders.to_scalar_dict()` returns `dict[str | bytes | bool | int
| float, str | bytes | bool | int | float]` — a direct, lossless
conversion (no logging); it only raises `ValueError` on a genuine decode
error of a stored field.
Header decode errors from known-but-invalid or unknown semantic kinds
surface as `ValueError`. `SendMessage(id=...)` (mapped to `u128`) is
part of the same binding update.
## Minor Fix
The Python test compose setup starts the server with fresh default root
credentials, binds HTTP/TCP/QUIC addresses explicitly, and uses `iggy
ping` for the healthcheck instead of the HTTP stats endpoint.
> **TODO:** make pre-commit and other scripts compatible with other
shells (e.g. zsh); commands like `mapfile` don't work in some shells.
## Discussion Notes
Choices from the issue discussion, updated to the final implementation:
- Python supports **both** a plain `dict` API for the common case
**and** explicit `HeaderKey`/`HeaderValue` wrapper classes; the two can
be mixed per entry.
- Integers are stored in the narrowest lossless kind (unsigned for
non-negative, signed for negative); only values beyond ±128-bit raise
`ValueError`.
- Floats select `Float32`/`Float64` by exact representability instead of
always `Float64`.
- Non-string keys are now **supported** (any plain scalar or
`HeaderKey`), instead of being rejected.
- `ReceiveMessage.user_headers()` returns `None` for no headers.
- `origin_timestamp()` is exposed on received messages.
## API Usage
Sending — plain scalars (common case)
```python
from apache_iggy import SendMessage as Message
# Keys and values are plain Python scalars; each is converted losslessly.
Message(
"order-created",
user_headers={
"content-type": "application/json", # -> String
"trace-blob": b"\x00\x01", # -> Raw
"retryable": False, # -> Bool
"attempt": 3, # -> UnsignedInt8 (smallest
fit)
"created-at-ms": 1_700_000_000_000, # -> UnsignedInt64
"score": 1.25, # -> Float32 (exactly
representable)
},
id=42, # optional custom message id (u128)
)
```
### Sending — explicit typed kinds
```python
from apache_iggy import HeaderKey, HeaderValue
from apache_iggy import SendMessage as Message
Message(
"order-created",
user_headers={
HeaderKey.String("schema-version"): HeaderValue.UnsignedInt16(1),
HeaderKey.UnsignedInt32(7): HeaderValue.String("order-id"), #
non-string key
},
)
```
### Sending — mixed typed + plain in one dict
```python
Message(
"order-created",
user_headers={
HeaderKey.String("typed"): HeaderValue.Int32(-5),
"plain-key": HeaderValue.Bool(True),
HeaderKey.String("plain-val"): "hello",
"both-plain": 9,
},
)
```
### Receiving
```python
message = polled_messages[0]
message.origin_timestamp() # int
headers = message.user_headers() # UserHeaders | None
if headers is not None:
# Typed access (full wire-type fidelity):
value = headers.get(HeaderKey.String("schema-version"))
if isinstance(value, HeaderValue.UnsignedInt16):
print(value.value) # 1
# Convenient plain form (lossless):
plain = headers.to_scalar_dict() # dict[str|bytes|bool|int|float,
str|bytes|bool|int|float]
print(plain["schema-version"]) # 1
```
### Validation / errors
```python
Message("p", user_headers={"k": 2**128}) # ValueError: 128-bit
range
Message("p", user_headers={"k": HeaderValue.Int8(300)}) # OverflowError
Message("p", user_headers={"k": HeaderValue.Float32(1e40)}) # ValueError:
32-bit float
Message("p", user_headers={object(): "v"}) # ValueError: keys must
be str/bytes/bool/int/float/HeaderKey
```
## Local Execution
- Passed: `cargo fmt --check --manifest-path foreign/python/Cargo.toml`
- Passed: `cargo check --manifest-path foreign/python/Cargo.toml`
- Passed: `cargo test --manifest-path foreign/python/Cargo.toml`
- Passed: `uv run --extra dev ruff check
tests/test_message_operations.py tests/test_consumer_group.py`
- Passed: `.venv/bin/python -m pytest
tests/test_message_operations.py::TestMessageOperations::test_invalid_user_headers_are_rejected
-q`
- Successfully run: `example/message-headers/typed-headers/consumer.py`,
`example/message-headers/typed-headers/producer.py`,
`example/message-headers/plain-headers/consumer.py`,
`example/message-headers/plain-headers/producer.py`
## AI Usage
Codex was used to inspect the existing Python, Rust, Node, and Go SDK
behavior, implement the Python binding changes, add tests. All the
modification was reviewed carefully by the human.
---------
Co-authored-by: Rimuksh Kansal <[email protected]>
Co-authored-by: Hubert Gruszecki <[email protected]>
---
core/sdk/src/prelude.rs | 6 +-
examples/python/README.md | 23 +
examples/python/basic/producer.py | 50 +-
examples/python/getting-started/producer.py | 50 +-
examples/python/message-headers/common.py | 240 +++++++
.../message-headers/plain-headers/consumer.py | 77 ++
.../message-headers/plain-headers/producer.py | 59 ++
.../message-headers/typed-headers/consumer.py | 77 ++
.../message-headers/typed-headers/producer.py | 64 ++
examples/python/pyproject.toml | 5 +
foreign/python/Cargo.toml | 1 +
foreign/python/Dockerfile.test | 2 +-
foreign/python/apache_iggy.pyi | 397 ++++++++++-
foreign/python/docker-compose.test.yml | 11 +-
foreign/python/src/lib.rs | 5 +
foreign/python/src/receive_message.rs | 26 +
foreign/python/src/send_message.rs | 41 +-
foreign/python/src/user_headers.rs | 791 +++++++++++++++++++++
foreign/python/tests/test_consumer_group.py | 98 +++
foreign/python/tests/test_message_operations.py | 402 ++++++++++-
20 files changed, 2345 insertions(+), 80 deletions(-)
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index 34f8d1484..d364c08ea 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -51,9 +51,9 @@ pub use iggy_common::{
Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, CacheMetrics,
CacheMetricsKey, ClientError,
ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole,
ClusterNodeStatus,
CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails,
ConsumerGroupMember,
- ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind,
HeaderValue,
- HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier,
IdentityInfo,
- IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView,
IggyMessage,
+ ConsumerKind, 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,
diff --git a/examples/python/README.md b/examples/python/README.md
index 9bf943b75..7e5da180d 100644
--- a/examples/python/README.md
+++ b/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:
diff --git a/examples/python/basic/producer.py
b/examples/python/basic/producer.py
index b04ca8fc8..aaa91fc46 100644
--- a/examples/python/basic/producer.py
+++ b/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.")
diff --git a/examples/python/getting-started/producer.py
b/examples/python/getting-started/producer.py
index c05c0317c..642399edb 100755
--- a/examples/python/getting-started/producer.py
+++ b/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.")
diff --git a/examples/python/message-headers/common.py
b/examples/python/message-headers/common.py
new file mode 100644
index 000000000..79640b15e
--- /dev/null
+++ b/examples/python/message-headers/common.py
@@ -0,0 +1,240 @@
+# 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 json
+import secrets
+import time
+from collections.abc import Callable, Iterator
+from dataclasses import dataclass
+from enum import Enum
+
+from apache_iggy import (
+ HeaderKey,
+ HeaderValue,
+ IggyClient,
+ PollingStrategy,
+ ReceiveMessage,
+ StreamDetails,
+ TopicDetails,
+)
+from apache_iggy import SendMessage as Message
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+PlainHeaderValue = str | bytes | bool | int | float
+PlainHeaders = dict[str, PlainHeaderValue]
+TypedHeaders = dict[HeaderKey, HeaderValue]
+HeadersBuilder = Callable[["Order"], PlainHeaders | TypedHeaders]
+MessageHandler = Callable[[ReceiveMessage], None]
+
+
+class OrderType(str, Enum):
+ CREATED = "OrderCreated"
+ CONFIRMED = "OrderConfirmed"
+ REJECTED = "OrderRejected"
+
+
+@dataclass(frozen=True, slots=True)
+class ArgNamespace:
+ connection_string: str
+
+
+@dataclass(frozen=True, slots=True)
+class Order:
+ order_type: OrderType
+ payload: str
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy client, e.g. "
+ "'iggy+tcp://iggy:[email protected]:8090'"
+ ),
+ default="iggy+tcp://iggy:[email protected]:8090",
+ nargs="?",
+ type=str,
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+def generate_orders() -> Iterator[Order]:
+ order_id = 0
+ while True:
+ order_id += 1
+ group_id = (order_id + 2) // 3
+ match order_id % 3:
+ case 1:
+ payload = {
+ "orderId": f"order-{group_id}",
+ "customerId": f"customer-{secrets.randbelow(100)}",
+ "amount": secrets.randbelow(10000) + 1,
+ }
+ yield Order(OrderType.CREATED, json.dumps(payload))
+ case 2:
+ payload = {
+ "orderId": f"order-{group_id}",
+ "timestamp": int(time.time() * 1000),
+ }
+ yield Order(OrderType.CONFIRMED, json.dumps(payload))
+ case _:
+ payload = {
+ "orderId": f"order-{group_id}",
+ "reason": "Insufficient balance",
+ }
+ yield Order(OrderType.REJECTED, json.dumps(payload))
+
+
+async def connect(connection_string: str) -> IggyClient:
+ client = IggyClient.from_connection_string(connection_string)
+ logger.info("Connecting to Iggy")
+ await client.connect()
+ logger.info("Connected")
+ return client
+
+
+async def init_system(client: IggyClient) -> None:
+ 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, build_headers: HeadersBuilder)
-> None:
+ interval = 0.5
+ logger.info(
+ f"Messages will be sent to stream: {STREAM_NAME}, "
+ f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} "
+ f"with interval {interval * 1000} ms."
+ )
+ orders = generate_orders()
+ sent_batches = 0
+
+ while sent_batches < BATCHES_LIMIT:
+ messages: list[Message] = []
+ for order in (next(orders) for _ in range(MESSAGES_PER_BATCH)):
+ headers = build_headers(order)
+ messages.append(Message(order.payload, user_headers=headers))
+ logger.info(
+ f"Prepared {order.order_type} with headers:
{format_headers(headers)}"
+ )
+
+ try:
+ await client.send_messages(
+ stream=STREAM_NAME,
+ topic=TOPIC_NAME,
+ partitioning=PARTITION_ID,
+ messages=messages,
+ )
+ sent_batches += 1
+ logger.info(f"Sent {len(messages)} message(s).")
+ 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 {sent_batches} batches of messages, exiting.")
+
+
+async def consume_messages(client: IggyClient, handle_message: MessageHandler)
-> None:
+ interval = 0.5
+ logger.info(
+ f"Messages will be consumed from stream: {STREAM_NAME}, "
+ f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} "
+ f"with interval {interval * 1000} ms."
+ )
+ consumed_batches = 0
+
+ while 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
+
+ for message in polled_messages:
+ handle_message(message)
+
+ consumed_batches += 1
+ logger.info(f"Consumed {len(polled_messages)} message(s).")
+ await asyncio.sleep(interval)
+ except Exception as error:
+ logger.exception(f"Exception occurred while consuming messages:
{error}")
+ break
+
+ logger.info(f"Consumed {consumed_batches} batches of messages, exiting.")
+
+
+def log_order(message_type: OrderType | None, payload: object) -> None:
+ match message_type:
+ case OrderType.CREATED:
+ logger.info(f"Order Created: {payload}")
+ case OrderType.CONFIRMED:
+ logger.info(f"Order Confirmed: {payload}")
+ case OrderType.REJECTED:
+ logger.info(f"Order Rejected: {payload}")
+ case _:
+ logger.warning(f"Received unknown message type: {message_type}")
+
+
+def format_headers(headers: dict) -> dict[str, str]:
+ formatted: dict[str, str] = {}
+ for key, value in headers.items():
+ formatted_key = repr(key) if isinstance(key, HeaderKey) else str(key)
+ if isinstance(value, bytes):
+ formatted[formatted_key] = f"bytes({value.hex()})"
+ elif isinstance(value, HeaderValue):
+ formatted[formatted_key] = repr(value)
+ else:
+ formatted[formatted_key] = f"{value!r} ({type(value).__name__})"
+ return formatted
diff --git a/examples/python/message-headers/plain-headers/consumer.py
b/examples/python/message-headers/plain-headers/consumer.py
new file mode 100644
index 000000000..92874b022
--- /dev/null
+++ b/examples/python/message-headers/plain-headers/consumer.py
@@ -0,0 +1,77 @@
+# 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 asyncio
+import json
+import sys
+from pathlib import Path
+
+sys.path.append(str(Path(__file__).resolve().parent.parent))
+
+from apache_iggy import ReceiveMessage # noqa: E402
+from common import ( # noqa: E402
+ OrderType,
+ PlainHeaderValue,
+ connect,
+ consume_messages,
+ format_headers,
+ log_order,
+ parse_args,
+)
+from loguru import logger # noqa: E402
+
+
+def handle_message(message: ReceiveMessage) -> None:
+ payload = json.loads(message.payload().decode("utf-8"))
+ headers = message.user_headers()
+
+ logger.info(
+ f"Handling message at offset {message.offset()} "
+ f"with origin timestamp {message.origin_timestamp()}."
+ )
+
+ scalar_headers: dict[PlainHeaderValue, PlainHeaderValue] = {}
+ if headers is not None:
+ # `to_scalar_dict` converts the typed headers stored on the wire back
+ # into the convenient plain `dict[str, str | bytes | bool | int |
+ # float]` form.
+ scalar_headers = headers.to_scalar_dict()
+ logger.info(f"Plain headers: {format_headers(scalar_headers)}")
+
+ log_order(get_message_type(scalar_headers), payload)
+
+
+def get_message_type(
+ headers: dict[PlainHeaderValue, PlainHeaderValue],
+) -> OrderType | None:
+ message_type = headers.get("message-type")
+ if isinstance(message_type, str):
+ try:
+ return OrderType(message_type)
+ except ValueError:
+ logger.warning(f"Received unknown message type: {message_type}")
+ return None
+
+
+async def main() -> None:
+ args = parse_args()
+ client = await connect(args.connection_string)
+ await consume_messages(client, handle_message)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/message-headers/plain-headers/producer.py
b/examples/python/message-headers/plain-headers/producer.py
new file mode 100644
index 000000000..397efe2ab
--- /dev/null
+++ b/examples/python/message-headers/plain-headers/producer.py
@@ -0,0 +1,59 @@
+# 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 asyncio
+import sys
+import time
+from pathlib import Path
+
+sys.path.append(str(Path(__file__).resolve().parent.parent))
+
+from common import ( # noqa: E402
+ Order,
+ OrderType,
+ PlainHeaders,
+ connect,
+ init_system,
+ parse_args,
+ produce_messages,
+)
+
+
+def build_plain_headers(order: Order) -> PlainHeaders:
+ # The plain `dict[str, str | bytes | bool | int | float]` form is the
+ # easiest way to attach headers: the SDK infers a wire type from each
+ # Python scalar and translates it into the typed form on the wire.
+ return {
+ "message-type": order.order_type,
+ "content-type": "application/json",
+ "schema-version": 1,
+ "created-at-ms": int(time.time() * 1000),
+ "retryable": order.order_type == OrderType.REJECTED,
+ "priority-score": 0.5,
+ "trace-bin": order.payload.encode(),
+ }
+
+
+async def main() -> None:
+ args = parse_args()
+ client = await connect(args.connection_string)
+ await init_system(client)
+ await produce_messages(client, build_plain_headers)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/message-headers/typed-headers/consumer.py
b/examples/python/message-headers/typed-headers/consumer.py
new file mode 100644
index 000000000..e24d531b2
--- /dev/null
+++ b/examples/python/message-headers/typed-headers/consumer.py
@@ -0,0 +1,77 @@
+# 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 asyncio
+import json
+import sys
+from pathlib import Path
+
+sys.path.append(str(Path(__file__).resolve().parent.parent))
+
+from apache_iggy import ( # noqa: E402
+ HeaderKey,
+ HeaderValue,
+ ReceiveMessage,
+ UserHeaders,
+)
+from common import ( # noqa: E402
+ OrderType,
+ connect,
+ consume_messages,
+ format_headers,
+ log_order,
+ parse_args,
+)
+from loguru import logger # noqa: E402
+
+
+def handle_message(message: ReceiveMessage) -> None:
+ payload = json.loads(message.payload().decode("utf-8"))
+ # `user_headers()` returns the explicitly typed `UserHeaders` mapping
+ # (a dict subclass) or None when the message carries no headers.
+ headers = message.user_headers()
+
+ logger.info(
+ f"Handling message at offset {message.offset()} "
+ f"with origin timestamp {message.origin_timestamp()}."
+ )
+ if headers is not None:
+ logger.info(f"Headers: {format_headers(headers)}")
+
+ log_order(get_message_type(headers), payload)
+
+
+def get_message_type(headers: UserHeaders | None) -> OrderType | None:
+ if headers is None:
+ return None
+ message_type = headers.get(HeaderKey.String("message-type"))
+ if isinstance(message_type, HeaderValue.String):
+ try:
+ return OrderType(message_type.value)
+ except ValueError:
+ logger.warning(f"Received unknown message type:
{message_type.value}")
+ return None
+
+
+async def main() -> None:
+ args = parse_args()
+ client = await connect(args.connection_string)
+ await consume_messages(client, handle_message)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/message-headers/typed-headers/producer.py
b/examples/python/message-headers/typed-headers/producer.py
new file mode 100644
index 000000000..07dac2d47
--- /dev/null
+++ b/examples/python/message-headers/typed-headers/producer.py
@@ -0,0 +1,64 @@
+# 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 asyncio
+import sys
+import time
+from pathlib import Path
+
+sys.path.append(str(Path(__file__).resolve().parent.parent))
+
+from apache_iggy import HeaderKey, HeaderValue # noqa: E402
+from common import ( # noqa: E402
+ Order,
+ OrderType,
+ TypedHeaders,
+ connect,
+ init_system,
+ parse_args,
+ produce_messages,
+)
+
+
+def build_typed_headers(order: Order) -> TypedHeaders:
+ # `HeaderKey`/`HeaderValue` preserve an explicit wire type. Use this form
+ # when you need control over the exact width/kind of a header (e.g. a
+ # 16-bit unsigned int) rather than letting the SDK infer it.
+ return {
+ HeaderKey.String("message-type"): HeaderValue.String(order.order_type),
+ HeaderKey.String("content-type"):
HeaderValue.String("application/json"),
+ HeaderKey.String("schema-version"): HeaderValue.UnsignedInt16(1),
+ HeaderKey.String("created-at-ms"): HeaderValue.UnsignedInt64(
+ int(time.time() * 1000)
+ ),
+ HeaderKey.String("retryable"): HeaderValue.Bool(
+ order.order_type == OrderType.REJECTED
+ ),
+ HeaderKey.String("priority-score"): HeaderValue.Float32(0.5),
+ HeaderKey.String("trace-bin"): HeaderValue.Raw(order.payload.encode()),
+ }
+
+
+async def main() -> None:
+ args = parse_args()
+ client = await connect(args.connection_string)
+ await init_system(client)
+ await produce_messages(client, build_typed_headers)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/pyproject.toml b/examples/python/pyproject.toml
index ab6f27631..f80dfc49b 100644
--- a/examples/python/pyproject.toml
+++ b/examples/python/pyproject.toml
@@ -54,3 +54,8 @@ exclude-newer = "7 days"
project-includes = [
"**/*.py*",
]
+search-path = [
+ "message-headers",
+ "message-headers/plain-headers",
+ "message-headers/typed-headers",
+]
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index 612cec38a..d408c0303 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -38,6 +38,7 @@ doc = false
bytes = "1.12.1"
futures = "0.3.33"
iggy = { path = "../../core/sdk", version = "0.10.3-edge.3" }
+paste = "1"
pyo3 = "0.29.0"
pyo3-async-runtimes = { version = "0.29.0", features = [
"attributes",
diff --git a/foreign/python/Dockerfile.test b/foreign/python/Dockerfile.test
index 777b7e98a..ee7b17aea 100644
--- a/foreign/python/Dockerfile.test
+++ b/foreign/python/Dockerfile.test
@@ -59,7 +59,7 @@ COPY foreign/python/NOTICE ./foreign/python/
COPY core/ /workspace/core/
# Copy Python SDK source (changes more frequently)
-COPY foreign/python/src/ ./src/
+COPY foreign/python/src/ ./foreign/python/src/
# Install dependencies and build extension using native uv workflow
WORKDIR /workspace/foreign/python
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index a6bb3035b..0ce240163 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -32,6 +32,8 @@ __all__ = [
"ConsumerGroup",
"ConsumerGroupDetails",
"ConsumerGroupMember",
+ "HeaderKey",
+ "HeaderValue",
"GlobalPermissions",
"IggyClient",
"IggyConsumer",
@@ -47,6 +49,7 @@ __all__ = [
"UserInfo",
"UserInfoDetails",
"UserStatus",
+ "UserHeaders",
]
class AutoCommit:
@@ -311,6 +314,356 @@ class ConsumerGroupMember:
Gets the collection of partitions the consumer group member is
consuming.
"""
+class HeaderKey:
+ r"""
+ Typed key for an Iggy user header.
+
+ Use these constructors when the header key must preserve an explicit
+ wire type instead of using the common string-key dictionary form.
+ """
+ def __hash__(self) -> builtins.int: ...
+ def __richcmp__(self, other: typing.Any, op: int) -> typing.Any: ...
+ def __repr__(self) -> builtins.str: ...
+ @typing.final
+ class Raw(HeaderKey):
+ r"""
+ Raw bytes key. The byte length must be 1..=255.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> bytes: ...
+ def __new__(cls, value: bytes) -> HeaderKey.Raw: ...
+
+ @typing.final
+ class String(HeaderKey):
+ r"""
+ UTF-8 string key. The encoded byte length must be 1..=255.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.str: ...
+ def __new__(cls, value: builtins.str) -> HeaderKey.String: ...
+
+ @typing.final
+ class Bool(HeaderKey):
+ r"""
+ Boolean key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.bool: ...
+ def __new__(cls, value: builtins.bool) -> HeaderKey.Bool: ...
+
+ @typing.final
+ class Int8(HeaderKey):
+ r"""
+ Signed 8-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.Int8: ...
+
+ @typing.final
+ class Int16(HeaderKey):
+ r"""
+ Signed 16-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.Int16: ...
+
+ @typing.final
+ class Int32(HeaderKey):
+ r"""
+ Signed 32-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.Int32: ...
+
+ @typing.final
+ class Int64(HeaderKey):
+ r"""
+ Signed 64-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.Int64: ...
+
+ @typing.final
+ class Int128(HeaderKey):
+ r"""
+ Signed 128-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.Int128: ...
+
+ @typing.final
+ class UnsignedInt8(HeaderKey):
+ r"""
+ Unsigned 8-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.UnsignedInt8: ...
+
+ @typing.final
+ class UnsignedInt16(HeaderKey):
+ r"""
+ Unsigned 16-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.UnsignedInt16: ...
+
+ @typing.final
+ class UnsignedInt32(HeaderKey):
+ r"""
+ Unsigned 32-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.UnsignedInt32: ...
+
+ @typing.final
+ class UnsignedInt64(HeaderKey):
+ r"""
+ Unsigned 64-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.UnsignedInt64: ...
+
+ @typing.final
+ class UnsignedInt128(HeaderKey):
+ r"""
+ Unsigned 128-bit integer key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderKey.UnsignedInt128: ...
+
+ @typing.final
+ class Float32(HeaderKey):
+ r"""
+ 32-bit floating point key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.float: ...
+ def __new__(cls, value: builtins.float) -> HeaderKey.Float32: ...
+
+ @typing.final
+ class Float64(HeaderKey):
+ r"""
+ 64-bit floating point key.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.float: ...
+ def __new__(cls, value: builtins.float) -> HeaderKey.Float64: ...
+
+class HeaderValue:
+ r"""
+ Typed value for an Iggy user header.
+
+ Use these constructors when the header value must preserve an explicit
+ wire type instead of using the common Python scalar dictionary form.
+ """
+ def __hash__(self) -> builtins.int: ...
+ def __richcmp__(self, other: typing.Any, op: int) -> typing.Any: ...
+ def __repr__(self) -> builtins.str: ...
+ @typing.final
+ class Raw(HeaderValue):
+ r"""
+ Raw bytes value. The byte length must be 1..=255.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> bytes: ...
+ def __new__(cls, value: bytes) -> HeaderValue.Raw: ...
+
+ @typing.final
+ class String(HeaderValue):
+ r"""
+ UTF-8 string value. The encoded byte length must be 1..=255.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.str: ...
+ def __new__(cls, value: builtins.str) -> HeaderValue.String: ...
+
+ @typing.final
+ class Bool(HeaderValue):
+ r"""
+ Boolean value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.bool: ...
+ def __new__(cls, value: builtins.bool) -> HeaderValue.Bool: ...
+
+ @typing.final
+ class Int8(HeaderValue):
+ r"""
+ Signed 8-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.Int8: ...
+
+ @typing.final
+ class Int16(HeaderValue):
+ r"""
+ Signed 16-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.Int16: ...
+
+ @typing.final
+ class Int32(HeaderValue):
+ r"""
+ Signed 32-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.Int32: ...
+
+ @typing.final
+ class Int64(HeaderValue):
+ r"""
+ Signed 64-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.Int64: ...
+
+ @typing.final
+ class Int128(HeaderValue):
+ r"""
+ Signed 128-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.Int128: ...
+
+ @typing.final
+ class UnsignedInt8(HeaderValue):
+ r"""
+ Unsigned 8-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.UnsignedInt8: ...
+
+ @typing.final
+ class UnsignedInt16(HeaderValue):
+ r"""
+ Unsigned 16-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.UnsignedInt16: ...
+
+ @typing.final
+ class UnsignedInt32(HeaderValue):
+ r"""
+ Unsigned 32-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.UnsignedInt32: ...
+
+ @typing.final
+ class UnsignedInt64(HeaderValue):
+ r"""
+ Unsigned 64-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.UnsignedInt64: ...
+
+ @typing.final
+ class UnsignedInt128(HeaderValue):
+ r"""
+ Unsigned 128-bit integer value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.int: ...
+ def __new__(cls, value: builtins.int) -> HeaderValue.UnsignedInt128:
...
+
+ @typing.final
+ class Float32(HeaderValue):
+ r"""
+ 32-bit floating point value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.float: ...
+ def __new__(cls, value: builtins.float) -> HeaderValue.Float32: ...
+
+ @typing.final
+ class Float64(HeaderValue):
+ r"""
+ 64-bit floating point value.
+ """
+
+ __match_args__ = ("value",)
+ @property
+ def value(self) -> builtins.float: ...
+ def __new__(cls, value: builtins.float) -> HeaderValue.Float64: ...
+
@typing.final
class GlobalPermissions:
r"""
@@ -1058,6 +1411,11 @@ class ReceiveMessage:
Retrieves the timestamp of the received message.
The timestamp represents the time of the message within its topic.
"""
+ def origin_timestamp(self) -> builtins.int:
+ r"""
+ Retrieves the origin timestamp of the received message.
+ The origin timestamp represents when the message was originally
created.
+ """
def id(self) -> builtins.int:
r"""
Retrieves the id of the received message.
@@ -1077,6 +1435,10 @@ class ReceiveMessage:
r"""
Retrieves the partition this message belongs to.
"""
+ def user_headers(self) -> UserHeaders | None:
+ r"""
+ Retrieves user headers attached to the received message.
+ """
@typing.final
class SendMessage:
@@ -1085,7 +1447,12 @@ class SendMessage:
This class wraps a Rust message meant for sending, facilitating
the creation of such messages from Python and their subsequent use in Rust.
"""
- def __new__(cls, data: builtins.str | bytes) -> SendMessage:
+ def __new__(
+ cls,
+ data: builtins.str | bytes,
+ user_headers: dict | None = None,
+ id: builtins.int | None = None,
+ ) -> SendMessage:
r"""
Constructs a new `SendMessage` instance from a string or bytes.
This method allows for the creation of a `SendMessage` instance
@@ -1362,3 +1729,31 @@ class UserStatus(enum.Enum):
r"""
The user account is inactive and cannot be used.
"""
+
+class UserHeaders(dict):
+ r"""
+ User headers dictionary returned by `ReceiveMessage.user_headers`.
+
+ This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping
+ operations work) that additionally exposes `to_scalar_dict` for the
convenient
+ scalar form.
+ """
+ def __new__(cls, mapping: dict | None = None) -> UserHeaders:
+ r"""
+ Wraps a mapping so its entries gain the `to_scalar_dict` helper.
+
+ Accepts a dict whose keys and values can each independently be
+ `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool |
+ int | float`). The inherited `dict` initializer copies the provided
+ mapping.
+ """
+ def to_scalar_dict(
+ self,
+ ) -> dict[str | bytes | bool | int | float, str | bytes | bool | int |
float]:
+ r"""
+ Converts these headers into the convenient plain dictionary form.
+
+ Returns an error if two distinct typed keys map to the same plain
+ Python scalar (e.g., `UnsignedInt8(1)` and `UnsignedInt16(1)` both
+ become `int(1)`), or if a stored field cannot be decoded.
+ """
diff --git a/foreign/python/docker-compose.test.yml
b/foreign/python/docker-compose.test.yml
index 92ffec6da..129b941bf 100644
--- a/foreign/python/docker-compose.test.yml
+++ b/foreign/python/docker-compose.test.yml
@@ -15,8 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-version: '3.8'
-
services:
iggy-server:
build:
@@ -24,15 +22,22 @@ services:
dockerfile: core/server/Dockerfile
args:
PROFILE: debug
+ command: ["--fresh", "--with-default-root-credentials"]
container_name: iggy-server-python-test
+ security_opt:
+ - seccomp:unconfined
networks:
- python-test-network
ports:
- "3000:3000"
- "8080:8080"
- "8090:8090"
+ environment:
+ - IGGY_HTTP_ADDRESS=0.0.0.0:3000
+ - IGGY_TCP_ADDRESS=0.0.0.0:8090
+ - IGGY_QUIC_ADDRESS=0.0.0.0:8080
healthcheck:
- test: [ "CMD", "curl", "-f", "http://localhost:3000/stats"; ]
+ test: [ "CMD", "iggy", "--tcp-server-address", "127.0.0.1:8090", "ping" ]
interval: 5s
timeout: 5s
retries: 12
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index bad1c7cf1..9bb2308f4 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -24,6 +24,7 @@ mod send_message;
mod stream;
mod topic;
mod user;
+mod user_headers;
use client::IggyClient;
use consumer::{
@@ -37,6 +38,7 @@ use send_message::SendMessage;
use stream::StreamDetails;
use topic::{Topic, TopicDetails};
use user::{UserInfo, UserInfoDetails, UserStatus};
+use user_headers::{HeaderKey, HeaderValue, UserHeaders};
/// A Python module implemented in Rust.
#[pymodule]
@@ -59,6 +61,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) ->
PyResult<()> {
m.add_class::<UserStatus>()?;
m.add_class::<UserInfo>()?;
m.add_class::<UserInfoDetails>()?;
+ m.add_class::<UserHeaders>()?;
+ m.add_class::<HeaderKey>()?;
+ m.add_class::<HeaderValue>()?;
m.add_class::<Permissions>()?;
m.add_class::<GlobalPermissions>()?;
m.add_class::<StreamPermissions>()?;
diff --git a/foreign/python/src/receive_message.rs
b/foreign/python/src/receive_message.rs
index aadf0772c..ecabb6cf2 100644
--- a/foreign/python/src/receive_message.rs
+++ b/foreign/python/src/receive_message.rs
@@ -16,10 +16,13 @@
// under the License.
use iggy::prelude::{IggyMessage as RustReceiveMessage, PollingStrategy as
RustPollingStrategy};
+use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum,
gen_stub_pymethods};
+use crate::user_headers::{UserHeaders, rust_user_headers_to_py};
+
/// A Python class representing a received message.
/// This class wraps a Rust message, allowing for access to its payload and
offset from Python.
#[pyclass]
@@ -50,6 +53,12 @@ impl ReceiveMessage {
self.inner.header.timestamp
}
+ /// Retrieves the origin timestamp of the received message.
+ /// The origin timestamp represents when the message was originally
created.
+ pub fn origin_timestamp(&self) -> u64 {
+ self.inner.header.origin_timestamp
+ }
+
/// Retrieves the id of the received message.
/// The id represents unique identifier of the message within its topic.
pub fn id(&self) -> u128 {
@@ -72,6 +81,23 @@ impl ReceiveMessage {
pub fn partition_id(&self) -> u32 {
self.partition_id
}
+
+ /// Retrieves user headers attached to the received message.
+ ///
+ /// Returns `None` when no headers are present or when the headers
+ /// on the wire are structurally malformed (those errors are logged
+ /// internally). Only known semantic decode errors raise `ValueError`.
+ #[gen_stub(override_return_type(type_repr = "UserHeaders | None"))]
+ pub fn user_headers<'a>(&self, py: Python<'a>) ->
PyResult<Option<Bound<'a, UserHeaders>>> {
+ let Some(headers) = self
+ .inner
+ .user_headers_map()
+ .map_err(|e| PyValueError::new_err(e.to_string()))?
+ else {
+ return Ok(None);
+ };
+ rust_user_headers_to_py(py, headers).map(Some)
+ }
}
#[derive(Clone, Copy)]
diff --git a/foreign/python/src/send_message.rs
b/foreign/python/src/send_message.rs
index 021125d7e..6c09eb313 100644
--- a/foreign/python/src/send_message.rs
+++ b/foreign/python/src/send_message.rs
@@ -17,12 +17,13 @@
use bytes::Bytes;
use iggy::prelude::{IggyMessage as RustIggyMessage, IggyMessageHeader};
-use pyo3::{prelude::*, types::PyBytes};
+use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
use pyo3_stub_gen::{
derive::{gen_stub_pyclass, gen_stub_pymethods},
impl_stub_type,
};
-use std::str::FromStr;
+
+use crate::user_headers::py_user_headers_to_rust;
/// A Python class representing a message to be sent.
/// This class wraps a Rust message meant for sending, facilitating
@@ -61,22 +62,36 @@ impl SendMessage {
/// This method allows for the creation of a `SendMessage` instance
/// directly from Python using the provided string or bytes data.
#[new]
- pub fn new(py: Python, data: PyMessagePayload) -> PyResult<Self> {
- let inner = match data {
- PyMessagePayload::String(data) => RustIggyMessage::from_str(&data)
- .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError,
_>(e.to_string()))?,
- PyMessagePayload::Bytes(data) => {
- let bytes = Bytes::from(data.extract::<Vec<u8>>(py)?);
- RustIggyMessage::builder()
- .payload(bytes)
- .build()
- .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError,
_>(e.to_string()))?
- }
+ #[pyo3(signature = (data, user_headers=None, id=None))]
+ pub fn new(
+ py: Python,
+ data: PyMessagePayload,
+ #[gen_stub(override_type(type_repr = "dict | None"))] user_headers:
Option<
+ &Bound<'_, PyAny>,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] id:
Option<u128>,
+ ) -> PyResult<Self> {
+ let payload = match data {
+ PyMessagePayload::String(data) => Bytes::from(data),
+ PyMessagePayload::Bytes(data) =>
Bytes::from(data.extract::<Vec<u8>>(py)?),
};
+ let user_headers = user_headers
+ .map(|headers| py_user_headers_to_rust(py, headers))
+ .transpose()?;
+ let inner = RustIggyMessage::builder()
+ .maybe_id(id)
+ .payload(payload)
+ .maybe_user_headers(user_headers)
+ .build()
+ .map_err(to_value_error)?;
Ok(Self { inner })
}
}
+fn to_value_error(error: impl ToString) -> PyErr {
+ PyValueError::new_err(error.to_string())
+}
+
#[derive(FromPyObject, IntoPyObject)]
pub enum PyMessagePayload {
#[pyo3(transparent, annotation = "str")]
diff --git a/foreign/python/src/user_headers.rs
b/foreign/python/src/user_headers.rs
new file mode 100644
index 000000000..3eaa843a3
--- /dev/null
+++ b/foreign/python/src/user_headers.rs
@@ -0,0 +1,791 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::BTreeMap;
+use std::collections::hash_map::DefaultHasher;
+use std::hash::{Hash, Hasher};
+
+use iggy::prelude::{
+ HeaderField, HeaderKey as RustHeaderKey, HeaderKind, HeaderValue as
RustHeaderValue,
+};
+use pyo3::PyClass;
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+use pyo3::pyclass::CompareOp;
+use pyo3::types::{PyBool, PyBytes, PyDict, PyFloat, PyInt, PyString, PyTuple};
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum,
gen_stub_pymethods};
+
+type RustUserHeaders = BTreeMap<RustHeaderKey, RustHeaderValue>;
+
+#[derive(PartialEq, Eq, Hash)]
+struct HeaderIdentity {
+ kind: u8,
+ value: Vec<u8>,
+}
+
+impl HeaderIdentity {
+ fn raw(value: Vec<u8>) -> Self {
+ Self {
+ kind: HeaderKind::Raw.as_code(),
+ value,
+ }
+ }
+
+ fn string(value: Vec<u8>) -> Self {
+ Self {
+ kind: HeaderKind::String.as_code(),
+ value,
+ }
+ }
+
+ fn bool(value: bool) -> Self {
+ Self {
+ kind: HeaderKind::Bool.as_code(),
+ value: vec![u8::from(value)],
+ }
+ }
+
+ fn int8(value: i8) -> Self {
+ Self {
+ kind: HeaderKind::Int8.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int16(value: i16) -> Self {
+ Self {
+ kind: HeaderKind::Int16.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int32(value: i32) -> Self {
+ Self {
+ kind: HeaderKind::Int32.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int64(value: i64) -> Self {
+ Self {
+ kind: HeaderKind::Int64.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int128(value: i128) -> Self {
+ Self {
+ kind: HeaderKind::Int128.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint8(value: u8) -> Self {
+ Self {
+ kind: HeaderKind::Uint8.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint16(value: u16) -> Self {
+ Self {
+ kind: HeaderKind::Uint16.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint32(value: u32) -> Self {
+ Self {
+ kind: HeaderKind::Uint32.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint64(value: u64) -> Self {
+ Self {
+ kind: HeaderKind::Uint64.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint128(value: u128) -> Self {
+ Self {
+ kind: HeaderKind::Uint128.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn float32(value: f32) -> Self {
+ Self {
+ kind: HeaderKind::Float32.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn float64(value: f64) -> Self {
+ Self {
+ kind: HeaderKind::Float64.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+}
+
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+/// Typed key for an Iggy user header.
+///
+/// Use these constructors when the header key must preserve an explicit
+/// wire type instead of using the common string-key dictionary form.
+pub enum HeaderKey {
+ /// Raw bytes key. The byte length must be 1..=255.
+ Raw { value: Py<PyBytes> },
+ /// UTF-8 string key. The encoded byte length must be 1..=255.
+ String { value: String },
+ /// Boolean key.
+ Bool { value: bool },
+ /// Signed 8-bit integer key.
+ Int8 { value: i8 },
+ /// Signed 16-bit integer key.
+ Int16 { value: i16 },
+ /// Signed 32-bit integer key.
+ Int32 { value: i32 },
+ /// Signed 64-bit integer key.
+ Int64 { value: i64 },
+ /// Signed 128-bit integer key.
+ Int128 { value: i128 },
+ /// Unsigned 8-bit integer key.
+ UnsignedInt8 { value: u8 },
+ /// Unsigned 16-bit integer key.
+ UnsignedInt16 { value: u16 },
+ /// Unsigned 32-bit integer key.
+ UnsignedInt32 { value: u32 },
+ /// Unsigned 64-bit integer key.
+ UnsignedInt64 { value: u64 },
+ /// Unsigned 128-bit integer key.
+ UnsignedInt128 { value: u128 },
+ /// 32-bit floating point key.
+ Float32 { value: f32 },
+ /// 64-bit floating point key.
+ Float64 { value: f64 },
+}
+
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+/// Typed value for an Iggy user header.
+///
+/// Use these constructors when the header value must preserve an explicit
+/// wire type instead of using the common Python scalar dictionary form.
+pub enum HeaderValue {
+ /// Raw bytes value. The byte length must be 1..=255.
+ Raw { value: Py<PyBytes> },
+ /// UTF-8 string value. The encoded byte length must be 1..=255.
+ String { value: String },
+ /// Boolean value.
+ Bool { value: bool },
+ /// Signed 8-bit integer value.
+ Int8 { value: i8 },
+ /// Signed 16-bit integer value.
+ Int16 { value: i16 },
+ /// Signed 32-bit integer value.
+ Int32 { value: i32 },
+ /// Signed 64-bit integer value.
+ Int64 { value: i64 },
+ /// Signed 128-bit integer value.
+ Int128 { value: i128 },
+ /// Unsigned 8-bit integer value.
+ UnsignedInt8 { value: u8 },
+ /// Unsigned 16-bit integer value.
+ UnsignedInt16 { value: u16 },
+ /// Unsigned 32-bit integer value.
+ UnsignedInt32 { value: u32 },
+ /// Unsigned 64-bit integer value.
+ UnsignedInt64 { value: u64 },
+ /// Unsigned 128-bit integer value.
+ UnsignedInt128 { value: u128 },
+ /// 32-bit floating point value.
+ Float32 { value: f32 },
+ /// 64-bit floating point value.
+ Float64 { value: f64 },
+}
+
+trait PyHeaderFieldToRust: PyClass + Sized {
+ type RustType;
+ fn from_pyheader(py: Python<'_>, value: &Self) -> PyResult<Self::RustType>;
+ fn from_plain(field: &Bound<'_, PyAny>) ->
PyResult<Option<Self::RustType>>;
+}
+
+fn py_header_field_to_rust<T: PyHeaderFieldToRust>(
+ py: Python<'_>,
+ field: &Bound<'_, PyAny>,
+) -> PyResult<T::RustType> {
+ if let Ok(header) = field.extract::<PyRef<'_, T>>() {
+ return T::from_pyheader(py, &header);
+ }
+ T::from_plain(field)?.ok_or_else(|| {
+ PyValueError::new_err(
+ "User header must be str, bytes, bool, int, float, HeaderKey, or
HeaderValue",
+ )
+ })
+}
+
+trait ToPlain {
+ fn to_plain<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>>;
+}
+
+/// Borrows a typed header ([`HeaderKey`] or [`HeaderValue`]) together with the
+/// GIL token for direct conversion to a plain Python scalar.
+struct HeaderToPlainRef<'py, 'value, T: ToPlain> {
+ py: Python<'py>,
+ value: &'value T,
+}
+
+impl<'py, T: ToPlain> TryFrom<HeaderToPlainRef<'py, '_, T>> for Bound<'py,
PyAny> {
+ type Error = PyErr;
+
+ fn try_from(inner: HeaderToPlainRef<'py, '_, T>) -> PyResult<Self> {
+ let HeaderToPlainRef { py, value } = inner;
+ value.to_plain(py)
+ }
+}
+
+macro_rules! header_type_impl {
+ ($py_ty:ident) => {
+ paste::paste! {
+ #[gen_stub_pymethods]
+ #[pymethods]
+ impl $py_ty {
+ pub fn __hash__(&self, py: Python<'_>) -> PyResult<isize> {
+ Ok(py_hash(header_identity_hash(self.identity(py)?)))
+ }
+
+ pub fn __richcmp__(
+ &self,
+ py: Python<'_>,
+ other: &Bound<'_, PyAny>,
+ op: CompareOp,
+ ) -> PyResult<Py<PyAny>> {
+ let identity = self.identity(py)?;
+ let Ok(other_header) = other.extract::<PyRef<'_,
$py_ty>>() else {
+ return match op {
+ CompareOp::Eq => {
+
Ok(false.into_pyobject(py)?.to_owned().into_any().unbind())
+ }
+ CompareOp::Ne => {
+
Ok(true.into_pyobject(py)?.to_owned().into_any().unbind())
+ }
+ _ => Ok(py.NotImplemented()),
+ };
+ };
+ let result = match op {
+ CompareOp::Eq => identity ==
other_header.identity(py)?,
+ CompareOp::Ne => identity !=
other_header.identity(py)?,
+ _ => return Ok(py.NotImplemented()),
+ };
+
Ok(result.into_pyobject(py)?.to_owned().into_any().unbind())
+ }
+
+ pub fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
+ self.repr(py)
+ }
+ }
+
+ struct [<Py $py_ty Ref>]<'py, 'value> {
+ py: Python<'py>,
+ value: &'value $py_ty,
+ }
+
+ impl TryFrom<[<Py $py_ty Ref>]<'_, '_>> for [<Rust $py_ty>] {
+ type Error = PyErr;
+
+ fn try_from(value: [<Py $py_ty Ref>]<'_, '_>) ->
PyResult<Self> {
+ let [<Py $py_ty Ref>] { py, value } = value;
+ match value {
+ $py_ty::Raw { value } => {
+ <[<Rust
$py_ty>]>::try_from(value.extract::<Vec<u8>>(py)?)
+ .map_err(to_value_error)
+ }
+ $py_ty::String { value } => {
+ <[<Rust
$py_ty>]>::try_from(value.as_str()).map_err(to_value_error)
+ }
+ $py_ty::Bool { value } => Ok((*value).into()),
+ $py_ty::Int8 { value } => Ok((*value).into()),
+ $py_ty::Int16 { value } => Ok((*value).into()),
+ $py_ty::Int32 { value } => Ok((*value).into()),
+ $py_ty::Int64 { value } => Ok((*value).into()),
+ $py_ty::Int128 { value } => Ok((*value).into()),
+ $py_ty::UnsignedInt8 { value } => Ok((*value).into()),
+ $py_ty::UnsignedInt16 { value } => Ok((*value).into()),
+ $py_ty::UnsignedInt32 { value } => Ok((*value).into()),
+ $py_ty::UnsignedInt64 { value } => Ok((*value).into()),
+ $py_ty::UnsignedInt128 { value } =>
Ok((*value).into()),
+ $py_ty::Float32 { value } => checked_float32(*value),
+ $py_ty::Float64 { value } => checked_float64(*value),
+ }
+ }
+ }
+
+ struct [<Rust $py_ty Ref>]<'py, 'value> {
+ py: Python<'py>,
+ value: &'value [<Rust $py_ty>],
+ }
+
+ impl<'py> TryFrom<[<Rust $py_ty Ref>]<'py, '_>> for Bound<'py,
$py_ty> {
+ type Error = PyErr;
+
+ fn try_from(value: [<Rust $py_ty Ref>]<'py, '_>) ->
PyResult<Self> {
+ let [<Rust $py_ty Ref>] { py, value } = value;
+ let value = match value.kind() {
+ HeaderKind::Raw => $py_ty::Raw {
+ value: PyBytes::new(py,
value.as_raw().map_err(to_value_error)?)
+ .unbind(),
+ },
+ HeaderKind::String => $py_ty::String {
+ value:
value.as_str().map_err(to_value_error)?.to_string(),
+ },
+ HeaderKind::Bool => $py_ty::Bool {
+ value: value.as_bool().map_err(to_value_error)?,
+ },
+ HeaderKind::Int8 => $py_ty::Int8 {
+ value: value.as_int8().map_err(to_value_error)?,
+ },
+ HeaderKind::Int16 => $py_ty::Int16 {
+ value: value.as_int16().map_err(to_value_error)?,
+ },
+ HeaderKind::Int32 => $py_ty::Int32 {
+ value: value.as_int32().map_err(to_value_error)?,
+ },
+ HeaderKind::Int64 => $py_ty::Int64 {
+ value: value.as_int64().map_err(to_value_error)?,
+ },
+ HeaderKind::Int128 => $py_ty::Int128 {
+ value: value.as_int128().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint8 => $py_ty::UnsignedInt8 {
+ value: value.as_uint8().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint16 => $py_ty::UnsignedInt16 {
+ value: value.as_uint16().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint32 => $py_ty::UnsignedInt32 {
+ value: value.as_uint32().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint64 => $py_ty::UnsignedInt64 {
+ value: value.as_uint64().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint128 => $py_ty::UnsignedInt128 {
+ value: value.as_uint128().map_err(to_value_error)?,
+ },
+ HeaderKind::Float32 => {
+ let v =
value.as_float32().map_err(to_value_error)?;
+ if !v.is_finite() {
+ return Err(PyValueError::new_err(
+ "User header with non-finite Float32
value",
+ ));
+ }
+ $py_ty::Float32 { value: v }
+ },
+ HeaderKind::Float64 => {
+ let v =
value.as_float64().map_err(to_value_error)?;
+ if !v.is_finite() {
+ return Err(PyValueError::new_err(
+ "User header with non-finite Float64
value",
+ ));
+ }
+ $py_ty::Float64 { value: v }
+ },
+ };
+ value.into_pyobject(py)
+ }
+ }
+
+ impl $py_ty {
+ fn identity(&self, py: Python<'_>) -> PyResult<HeaderIdentity>
{
+ match self {
+ $py_ty::Raw { value } => {
+
Ok(HeaderIdentity::raw(value.extract::<Vec<u8>>(py)?))
+ }
+ $py_ty::String { value } => {
+
Ok(HeaderIdentity::string(value.as_bytes().to_vec()))
+ }
+ $py_ty::Bool { value } =>
Ok(HeaderIdentity::bool(*value)),
+ $py_ty::Int8 { value } =>
Ok(HeaderIdentity::int8(*value)),
+ $py_ty::Int16 { value } =>
Ok(HeaderIdentity::int16(*value)),
+ $py_ty::Int32 { value } =>
Ok(HeaderIdentity::int32(*value)),
+ $py_ty::Int64 { value } =>
Ok(HeaderIdentity::int64(*value)),
+ $py_ty::Int128 { value } =>
Ok(HeaderIdentity::int128(*value)),
+ $py_ty::UnsignedInt8 { value } =>
Ok(HeaderIdentity::uint8(*value)),
+ $py_ty::UnsignedInt16 { value } =>
Ok(HeaderIdentity::uint16(*value)),
+ $py_ty::UnsignedInt32 { value } =>
Ok(HeaderIdentity::uint32(*value)),
+ $py_ty::UnsignedInt64 { value } =>
Ok(HeaderIdentity::uint64(*value)),
+ $py_ty::UnsignedInt128 { value } =>
Ok(HeaderIdentity::uint128(*value)),
+ $py_ty::Float32 { value } =>
Ok(HeaderIdentity::float32(*value)),
+ $py_ty::Float64 { value } =>
Ok(HeaderIdentity::float64(*value)),
+ }
+ }
+
+ fn repr(&self, py: Python<'_>) -> PyResult<String> {
+ match self {
+ $py_ty::Raw { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Raw({})"),
+ value.bind(py).repr()?.extract::<String>()?
+ )),
+ $py_ty::String { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".String({:?})"),
+ value
+ )),
+ $py_ty::Bool { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Bool({})"),
+ value
+ )),
+ $py_ty::Int8 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Int8({})"),
+ value
+ )),
+ $py_ty::Int16 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Int16({})"),
+ value
+ )),
+ $py_ty::Int32 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Int32({})"),
+ value
+ )),
+ $py_ty::Int64 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Int64({})"),
+ value
+ )),
+ $py_ty::Int128 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Int128({})"),
+ value
+ )),
+ $py_ty::UnsignedInt8 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".UnsignedInt8({})"),
+ value
+ )),
+ $py_ty::UnsignedInt16 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".UnsignedInt16({})"),
+ value
+ )),
+ $py_ty::UnsignedInt32 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".UnsignedInt32({})"),
+ value
+ )),
+ $py_ty::UnsignedInt64 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".UnsignedInt64({})"),
+ value
+ )),
+ $py_ty::UnsignedInt128 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".UnsignedInt128({})"),
+ value
+ )),
+ $py_ty::Float32 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Float32({:?})"),
+ value
+ )),
+ $py_ty::Float64 { value } => Ok(format!(
+ concat!(stringify!($py_ty), ".Float64({:?})"),
+ value
+ )),
+ }
+ }
+ }
+
+ impl PyHeaderFieldToRust for $py_ty {
+ type RustType = [<Rust $py_ty>];
+
+ fn from_pyheader(py: Python<'_>, value: &Self) ->
PyResult<[<Rust $py_ty>]> {
+ <[<Rust $py_ty>]>::try_from([<Py $py_ty Ref>] { py, value
})
+ }
+
+ fn from_plain(field: &Bound<'_, PyAny>) ->
PyResult<Option<[<Rust $py_ty>]>> {
+ plain_to_rust_header_field(field)
+ }
+ }
+
+ impl ToPlain for $py_ty {
+ fn to_plain<'py>(&self, py: Python<'py>) ->
PyResult<Bound<'py, PyAny>> {
+ match self {
+ Self::Raw { value } =>
Ok(value.bind(py).clone().into_any()),
+ Self::String { value } =>
Ok(value.clone().into_pyobject(py)?.into_any()),
+ Self::Bool { value } =>
Ok((*value).into_pyobject(py)?.to_owned().into_any()),
+ Self::Int8 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::Int16 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::Int32 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::Int64 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::Int128 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::UnsignedInt8 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::UnsignedInt16 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::UnsignedInt32 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::UnsignedInt64 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::UnsignedInt128 { value } =>
Ok((*value).into_pyobject(py)?.into_any()),
+ Self::Float32 { value } => {
+ if !value.is_finite() {
+ return Err(PyValueError::new_err(
+ "User header with non-finite Float32
value",
+ ));
+ }
+ Ok(f64::from(*value).into_pyobject(py)?.into_any())
+ },
+ Self::Float64 { value } => {
+ if !value.is_finite() {
+ return Err(PyValueError::new_err(
+ "User header with non-finite Float64
value",
+ ));
+ }
+ Ok((*value).into_pyobject(py)?.into_any())
+ },
+ }
+ }
+ }
+ }
+ };
+}
+
+header_type_impl!(HeaderKey);
+header_type_impl!(HeaderValue);
+
+pub(crate) fn py_user_headers_to_rust(
+ py: Python<'_>,
+ mapping: &Bound<'_, PyAny>,
+) -> PyResult<RustUserHeaders> {
+ let mut rust_headers = BTreeMap::new();
+ for item in mapping.call_method0("items")?.try_iter()? {
+ let item = item?;
+ let pair: &Bound<'_, PyTuple> = item.cast()?;
+ let key = pair.get_item(0)?;
+ let value = pair.get_item(1)?;
+ let key = py_header_field_to_rust::<HeaderKey>(py, &key)?;
+ let value = py_header_field_to_rust::<HeaderValue>(py, &value)?;
+ if rust_headers.insert(key, value).is_some() {
+ return Err(PyValueError::new_err(
+ "Duplicate user header key: each header key must be unique",
+ ));
+ }
+ }
+ Ok(rust_headers)
+}
+
+pub(crate) fn rust_user_headers_to_py<'a>(
+ py: Python<'a>,
+ headers: RustUserHeaders,
+) -> PyResult<Bound<'a, UserHeaders>> {
+ let result = Bound::new(py, UserHeaders)?;
+ let mapping = result.as_any();
+ for (key, value) in headers {
+ let key = Bound::<HeaderKey>::try_from(RustHeaderKeyRef { py, value:
&key })?;
+ let value = Bound::<HeaderValue>::try_from(RustHeaderValueRef { py,
value: &value })?;
+ mapping.set_item(key, value)?;
+ }
+ Ok(result)
+}
+
+/// User headers dictionary returned by `ReceiveMessage.user_headers`.
+///
+/// This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping
+/// operations work) that additionally exposes `to_scalar_dict` for the
convenient
+/// scalar form.
+#[gen_stub_pyclass]
+#[pyclass(extends=PyDict)]
+pub struct UserHeaders;
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl UserHeaders {
+ /// Wraps a mapping so its entries gain the `to_scalar_dict` helper.
+ ///
+ /// Accepts a dict whose keys and values can each independently be
+ /// `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool |
+ /// int | float`). The inherited `dict` initializer copies the provided
+ /// mapping.
+ #[new]
+ #[pyo3(signature = (mapping=None))]
+ pub fn new(
+ #[gen_stub(override_type(type_repr = "dict | None"))] mapping:
Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<Self> {
+ if let Some(mapping) = mapping {
+ let py = mapping.py();
+ py_user_headers_to_rust(py, mapping)?;
+ }
+ Ok(UserHeaders)
+ }
+
+ fn __setitem__(
+ slf: &Bound<'_, Self>,
+ key: &Bound<'_, PyAny>,
+ value: &Bound<'_, PyAny>,
+ ) -> PyResult<()> {
+ let py = key.py();
+ py_header_field_to_rust::<HeaderKey>(py, key)?;
+ py_header_field_to_rust::<HeaderValue>(py, value)?;
+ slf.as_any().cast::<PyDict>()?.set_item(key, value)?;
+ Ok(())
+ }
+
+ /// Converts these headers into the convenient plain dictionary form.
+ ///
+ /// Returns an error if two distinct typed keys map to the same plain
+ /// Python scalar (e.g., `UnsignedInt8(1)` and `UnsignedInt16(1)` both
+ /// become `int(1)`), or if a stored field cannot be decoded.
+ #[gen_stub(override_return_type(
+ type_repr = "dict[str | bytes | bool | int | float, str | bytes | bool
| int | float]"
+ ))]
+ pub fn to_scalar_dict<'a>(slf: &Bound<'a, Self>) -> PyResult<Bound<'a,
PyDict>> {
+ let py = slf.py();
+ let dict = slf.as_any().cast::<PyDict>()?;
+ let result = PyDict::new(py);
+ for (key, value) in dict.iter() {
+ let plain_key = py_header_to_plain::<HeaderKey>(py, &key)?;
+ let plain_value = py_header_to_plain::<HeaderValue>(py, &value)?;
+ if result.contains(&plain_key)? {
+ return Err(PyValueError::new_err(
+ "Distinct typed header keys produce the same plain Python
scalar; this conversion is lossy and cannot proceed",
+ ));
+ }
+ result.set_item(plain_key, plain_value)?;
+ }
+ Ok(result)
+ }
+}
+
+fn py_header_to_plain<'py, T: ToPlain + PyClass>(
+ py: Python<'py>,
+ any: &Bound<'py, PyAny>,
+) -> PyResult<Bound<'py, PyAny>> {
+ if let Ok(header) = any.extract::<PyRef<'_, T>>() {
+ return Bound::<PyAny>::try_from(HeaderToPlainRef {
+ py,
+ value: &*header,
+ });
+ }
+ if any.is_instance_of::<PyString>()
+ || any.is_instance_of::<PyBytes>()
+ || any.is_instance_of::<PyInt>()
+ || any.is_instance_of::<PyFloat>()
+ {
+ return Ok(any.clone());
+ }
+ Err(PyValueError::new_err(
+ "User header must be str, bytes, bool, int, float, HeaderKey, or
HeaderValue",
+ ))
+}
+
+/// Converts a plain Python scalar into a header field (used for both keys and
+/// values). Returns `Ok(None)` when the object is not a supported plain type
so
+/// the caller can raise a message naming keys vs values; genuine range/length
+/// errors are returned as `Err`.
+fn plain_to_rust_header_field<T>(obj: &Bound<'_, PyAny>) ->
PyResult<Option<HeaderField<T>>> {
+ if obj.is_instance_of::<PyBool>() {
+ return Ok(Some(obj.extract::<bool>()?.into()));
+ }
+ if obj.is_instance_of::<PyInt>() {
+ return int_to_rust_header_field(obj).map(Some);
+ }
+ if obj.is_instance_of::<PyFloat>() {
+ return float_to_rust_header_field(obj.extract::<f64>()?).map(Some);
+ }
+ if obj.is_instance_of::<PyString>() {
+ return HeaderField::try_from(obj.extract::<String>()?)
+ .map(Some)
+ .map_err(to_value_error);
+ }
+ if obj.is_instance_of::<PyBytes>() {
+ return HeaderField::try_from(obj.extract::<Vec<u8>>()?)
+ .map(Some)
+ .map_err(to_value_error);
+ }
+ Ok(None)
+}
+
+fn int_to_rust_header_field<T>(obj: &Bound<'_, PyAny>) ->
PyResult<HeaderField<T>> {
+ if let Ok(value) = obj.extract::<u8>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<u16>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<u32>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<u64>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<u128>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<i8>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<i16>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<i32>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<i64>() {
+ return Ok(value.into());
+ }
+ if let Ok(value) = obj.extract::<i128>() {
+ return Ok(value.into());
+ }
+ Err(PyValueError::new_err(
+ "User header int values must fit within the 128-bit range",
+ ))
+}
+
+fn float_to_rust_header_field<T>(value: f64) -> PyResult<HeaderField<T>> {
+ if !value.is_finite() {
+ return Err(PyValueError::new_err(
+ "User header float values must be finite",
+ ));
+ }
+ let narrowed = value as f32;
+ if f64::from(narrowed) == value {
+ checked_float32(narrowed)
+ } else {
+ Ok(value.into())
+ }
+}
+
+fn checked_float32<T>(value: f32) -> PyResult<HeaderField<T>> {
+ if !value.is_finite() {
+ return Err(PyValueError::new_err(
+ "Float32 header must be a finite value within the 32-bit float
range",
+ ));
+ }
+ Ok(value.into())
+}
+
+fn checked_float64<T>(value: f64) -> PyResult<HeaderField<T>> {
+ if !value.is_finite() {
+ return Err(PyValueError::new_err("Float64 header value must be
finite"));
+ }
+ Ok(value.into())
+}
+
+fn header_identity_hash(identity: HeaderIdentity) -> u64 {
+ let mut hasher = DefaultHasher::new();
+ identity.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn py_hash(hash: u64) -> isize {
+ let hash = hash as isize;
+ if hash == -1 { -2 } else { hash }
+}
+
+fn to_value_error(error: impl ToString) -> PyErr {
+ PyValueError::new_err(error.to_string())
+}
diff --git a/foreign/python/tests/test_consumer_group.py
b/foreign/python/tests/test_consumer_group.py
index acb126048..f8adf8262 100644
--- a/foreign/python/tests/test_consumer_group.py
+++ b/foreign/python/tests/test_consumer_group.py
@@ -1110,6 +1110,58 @@ class TestConsumerGroup:
assert received_messages == test_messages
+ @pytest.mark.asyncio
+ async def test_consume_messages_can_read_user_headers(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test consume_messages callback receives user headers."""
+ consumer_name = unique_name()
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ expected_headers: dict[str, str | bytes | bool | int | float] = {
+ "source": "callback",
+ "attempt": 1,
+ }
+ received_headers = []
+ shutdown_event = asyncio.Event()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ consumer = await iggy_client.consumer_group(
+ consumer_name,
+ stream_name,
+ topic_name,
+ partition_id,
+ PollingStrategy.Next(),
+ 10,
+ auto_commit=AutoCommit.Disabled(),
+ poll_interval=timedelta(milliseconds=25),
+ )
+
+ async def take(message: ReceiveMessage) -> None:
+ headers = message.user_headers()
+ assert headers is not None
+ received_headers.append(headers.to_scalar_dict())
+ shutdown_event.set()
+
+ async def send() -> None:
+ await iggy_client.send_messages(
+ stream_name,
+ topic_name,
+ partition_id,
+ [Message("callback headers", user_headers=expected_headers)],
+ )
+
+ await asyncio.gather(consumer.consume_messages(take, shutdown_event),
send())
+
+ assert received_headers == [expected_headers]
+
@pytest.mark.asyncio
async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
"""Test that the consumer group can iterate over messages."""
@@ -1152,6 +1204,52 @@ class TestConsumerGroup:
assert received_messages == test_messages
+ @pytest.mark.asyncio
+ async def test_iter_messages_can_read_user_headers(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test iter_messages yields messages with user headers."""
+ consumer_name = unique_name()
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ expected_headers: dict[str, str | bytes | bool | int | float] = {
+ "source": "iterator",
+ "attempt": 2,
+ }
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ consumer = await iggy_client.consumer_group(
+ consumer_name,
+ stream_name,
+ topic_name,
+ partition_id,
+ PollingStrategy.Next(),
+ 10,
+ auto_commit=AutoCommit.Disabled(),
+ poll_interval=timedelta(milliseconds=25),
+ )
+
+ await iggy_client.send_messages(
+ stream_name,
+ topic_name,
+ partition_id,
+ [Message("iterator headers", user_headers=expected_headers)],
+ )
+
+ iterator = consumer.iter_messages()
+ message = await asyncio.wait_for(iterator.__anext__(), timeout=5)
+
+ headers = message.user_headers()
+ assert headers is not None
+ assert headers.to_scalar_dict() == expected_headers
+
@pytest.mark.asyncio
async def test_iter_messages_with_first_reads_existing_messages(
self, iggy_client: IggyClient, unique_name
diff --git a/foreign/python/tests/test_message_operations.py
b/foreign/python/tests/test_message_operations.py
index da57637c0..6a1974bf8 100644
--- a/foreign/python/tests/test_message_operations.py
+++ b/foreign/python/tests/test_message_operations.py
@@ -20,7 +20,13 @@ import uuid
import pytest
-from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import (
+ HeaderKey,
+ HeaderValue,
+ IggyClient,
+ PollingStrategy,
+ UserHeaders,
+)
from apache_iggy import SendMessage as Message
@@ -136,8 +142,158 @@ class TestMessageOperations:
assert isinstance(msg.offset(), int) and msg.offset() >= 0
assert isinstance(msg.id(), int) and msg.id() > 0
assert isinstance(msg.timestamp(), int) and msg.timestamp() > 0
+ assert isinstance(msg.origin_timestamp(), int) and
msg.origin_timestamp() > 0
assert isinstance(msg.checksum(), int)
assert isinstance(msg.length(), int) and msg.length() > 0
+ assert msg.user_headers() is None
+
+ @pytest.mark.asyncio
+ async def test_message_user_headers_round_trip(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test plain user headers round-trip through the typed
representation."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ message_id = 123456789
+ user_headers = {
+ "content-type": "application/json",
+ "trace-blob": b"\x00\x01",
+ "is-retry": False,
+ "attempt": 3,
+ "score": 0.99,
+ }
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[
+ Message(
+ "header round trip",
+ user_headers=user_headers,
+ id=message_id,
+ )
+ ],
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Last(),
+ count=1,
+ auto_commit=True,
+ )
+
+ assert len(polled_messages) == 1
+ message = polled_messages[0]
+ assert message.id() == message_id
+ typed_headers = message.user_headers()
+ assert typed_headers is not None
+ assert typed_headers == {
+ HeaderKey.String("content-type"):
HeaderValue.String("application/json"),
+ HeaderKey.String("trace-blob"): HeaderValue.Raw(b"\x00\x01"),
+ HeaderKey.String("is-retry"): HeaderValue.Bool(False),
+ HeaderKey.String("attempt"): HeaderValue.UnsignedInt8(3),
+ HeaderKey.String("score"): HeaderValue.Float64(0.99),
+ }
+ assert typed_headers.to_scalar_dict() == user_headers
+ assert isinstance(message.origin_timestamp(), int)
+ assert message.origin_timestamp() > 0
+
+ @pytest.mark.asyncio
+ async def test_typed_message_user_headers_round_trip(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test typed user headers preserve explicit header kinds."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ user_headers: dict[HeaderKey, HeaderValue] = {
+ HeaderKey.UnsignedInt128(42): HeaderValue.UnsignedInt128(2**96),
+ HeaderKey.String("float32"): HeaderValue.Float32(1.25),
+ }
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message("typed headers", user_headers=user_headers)], #
pyright: ignore[reportArgumentType]
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Last(),
+ count=1,
+ auto_commit=True,
+ )
+
+ assert len(polled_messages) == 1
+ headers = polled_messages[0].user_headers()
+ assert headers == user_headers
+
+ @pytest.mark.asyncio
+ async def test_plain_scalars_pick_smallest_lossless_kind(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test plain ints/floats map to the narrowest lossless header kind."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ plain_headers = {
+ "small": 200,
+ "negative": -5,
+ "large": 2**63,
+ "huge": 2**96,
+ "exact-float": 1.25,
+ "wide-float": 0.1,
+ }
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message("plain scalars", user_headers=plain_headers)],
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Last(),
+ count=1,
+ auto_commit=True,
+ )
+
+ headers = polled_messages[0].user_headers()
+ assert headers is not None
+ assert headers == {
+ HeaderKey.String("small"): HeaderValue.UnsignedInt8(200),
+ HeaderKey.String("negative"): HeaderValue.Int8(-5),
+ HeaderKey.String("large"): HeaderValue.UnsignedInt64(2**63),
+ HeaderKey.String("huge"): HeaderValue.UnsignedInt128(2**96),
+ HeaderKey.String("exact-float"): HeaderValue.Float32(1.25),
+ HeaderKey.String("wide-float"): HeaderValue.Float64(0.1),
+ }
+ assert headers.to_scalar_dict() == plain_headers
@pytest.mark.asyncio
@pytest.mark.parametrize(
@@ -149,6 +305,250 @@ class TestMessageOperations:
with pytest.raises(ValueError, match="Invalid message payload length"):
Message(payload)
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("headers", "error"),
+ [
+ ({"": "value"}, "Invalid header value"),
+ ({"x" * 256: "value"}, "Invalid header value"),
+ ({"key": ""}, "Invalid header value"),
+ ({"key": b""}, "Invalid header value"),
+ ({"key": "x" * 256}, "Invalid header value"),
+ (
+ {object(): "value"},
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue",
+ ),
+ (
+ {"key": object()},
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue",
+ ),
+ ({"key": 2**128}, "128-bit range"),
+ ({"key": -(2**200)}, "128-bit range"),
+ ({"key": float("inf")}, "finite"),
+ ({"key": float("-inf")}, "finite"),
+ ({"key": float("nan")}, "finite"),
+ ],
+ )
+ async def test_invalid_user_headers_are_rejected(self, headers, error):
+ """Test invalid user header input raises ValueError."""
+ with pytest.raises(ValueError, match=error):
+ Message("payload", user_headers=headers)
+
+ def test_duplicate_user_header_keys_are_rejected(self):
+ """Test a message with duplicate user header keys is rejected."""
+ with pytest.raises(ValueError, match="Duplicate user header key"):
+ Message(
+ "payload",
+ user_headers={
+ HeaderKey.String("dup"): HeaderValue.String("first"),
+ "dup": "second",
+ },
+ )
+
+ def test_user_headers_over_100k_bytes_are_rejected(self):
+ """Test user headers exceeding the 100 KB limit are rejected."""
+ oversized_headers = {f"key-{index:05d}": "v" * 255 for index in
range(1000)}
+ with pytest.raises(ValueError, match="Too big headers payload"):
+ Message("payload", user_headers=oversized_headers)
+
+ def test_explicit_float32_out_of_range_is_rejected(self):
+ """Test an explicit Float32 whose value overflows f32 is rejected."""
+ with pytest.raises(ValueError, match="32-bit float"):
+ Message("payload", user_headers={"k": HeaderValue.Float32(1e40)})
+
+ @pytest.mark.parametrize(
+ "kind,val_str",
+ [
+ (HeaderValue.Float32, "inf"),
+ (HeaderValue.Float32, "-inf"),
+ (HeaderValue.Float32, "nan"),
+ (HeaderValue.Float64, "inf"),
+ (HeaderValue.Float64, "-inf"),
+ (HeaderValue.Float64, "nan"),
+ ],
+ )
+ def test_non_finite_typed_header_value_is_rejected(self, kind, val_str):
+ """Test typed Float32/Float64 header values reject non-finite
values."""
+ value = float(val_str)
+ with pytest.raises(ValueError, match="finite"):
+ UserHeaders({"k": kind(value)})
+
+ @pytest.mark.parametrize(
+ "kind,val_str",
+ [
+ (HeaderKey.Float32, "inf"),
+ (HeaderKey.Float32, "-inf"),
+ (HeaderKey.Float32, "nan"),
+ (HeaderKey.Float64, "inf"),
+ (HeaderKey.Float64, "-inf"),
+ (HeaderKey.Float64, "nan"),
+ ],
+ )
+ def test_non_finite_typed_header_key_is_rejected(self, kind, val_str):
+ """Test typed Float32/Float64 header keys reject non-finite values."""
+ value = float(val_str)
+ with pytest.raises(ValueError, match="finite"):
+ UserHeaders({kind(value): "v"})
+
+ def test_mixed_typed_and_plain_headers_can_be_constructed(self):
+ """Test each key/value pair is converted independently and can be
mixed."""
+ Message(
+ "payload",
+ user_headers={
+ HeaderKey.String("typed-key"): HeaderValue.UnsignedInt16(7),
+ HeaderKey.String("plain-value"): "still-a-string",
+ "plain-key": HeaderValue.Bool(True),
+ "fully-plain": 42,
+ },
+ )
+
+ def test_typed_user_headers_can_be_constructed(self):
+ """Test typed header keys and values cover the full header kind
surface."""
+ Message(
+ "payload",
+ user_headers={
+ HeaderKey.Raw(b"raw-key"): HeaderValue.Raw(b"raw-value"),
+ HeaderKey.String("string-key"):
HeaderValue.String("string-value"),
+ HeaderKey.Bool(True): HeaderValue.Bool(False),
+ HeaderKey.Int8(-8): HeaderValue.Int8(-7),
+ HeaderKey.Int16(-16): HeaderValue.Int16(-15),
+ HeaderKey.Int32(-32): HeaderValue.Int32(-31),
+ HeaderKey.Int64(-64): HeaderValue.Int64(-63),
+ HeaderKey.Int128(-(2**80)): HeaderValue.Int128(-(2**79)),
+ HeaderKey.UnsignedInt8(8): HeaderValue.UnsignedInt8(9),
+ HeaderKey.UnsignedInt16(16): HeaderValue.UnsignedInt16(17),
+ HeaderKey.UnsignedInt32(32): HeaderValue.UnsignedInt32(33),
+ HeaderKey.UnsignedInt64(64): HeaderValue.UnsignedInt64(65),
+ HeaderKey.UnsignedInt128(2**80):
HeaderValue.UnsignedInt128(2**79),
+ HeaderKey.Float32(1.25): HeaderValue.Float32(2.5),
+ HeaderKey.Float64(3.5): HeaderValue.Float64(4.75),
+ },
+ )
+
+ def test_plain_user_headers_convert_every_kind_losslessly(self):
+ """Test every header kind converts to a plain scalar without
logging."""
+ headers = UserHeaders(
+ {
+ HeaderKey.String("content-type"): HeaderValue.String(
+ "application/json"
+ ),
+ HeaderKey.String("trace-blob"): HeaderValue.Raw(b"\x00\x01"),
+ HeaderKey.String("is-retry"): HeaderValue.Bool(True),
+ HeaderKey.String("attempt"): HeaderValue.Int64(3),
+ HeaderKey.String("schema-version"):
HeaderValue.UnsignedInt16(1),
+ HeaderKey.String("big"): HeaderValue.UnsignedInt128(2**96),
+ HeaderKey.String("ratio"): HeaderValue.Float32(1.25),
+ HeaderKey.String("score"): HeaderValue.Float64(0.5),
+ }
+ )
+
+ plain = headers.to_scalar_dict()
+
+ assert plain == {
+ "content-type": "application/json",
+ "trace-blob": b"\x00\x01",
+ "is-retry": True,
+ "attempt": 3,
+ "schema-version": 1,
+ "big": 2**96,
+ "ratio": 1.25,
+ "score": 0.5,
+ }
+
+ def test_plain_user_headers_accepts_plain_dict(self):
+ """Test the plain dictionary form passes through unchanged."""
+ headers = {"content-type": "application/json", "attempt": 3}
+ assert UserHeaders(headers).to_scalar_dict() == headers
+
+ def test_plain_user_headers_preserve_non_string_keys(self):
+ """Test non-string typed keys convert back to their scalar Python
type."""
+ headers = UserHeaders(
+ {HeaderKey.UnsignedInt32(7): HeaderValue.String("order-id")}
+ )
+
+ plain = headers.to_scalar_dict()
+
+ assert plain == {7: "order-id"}
+
+ def test_to_scalar_dict_raises_on_colliding_keys(self):
+ """Test that typed keys mapping to the same plain scalar raise an
error."""
+ headers = UserHeaders(
+ {
+ HeaderKey.UnsignedInt8(1): HeaderValue.String("first"),
+ HeaderKey.UnsignedInt16(1): HeaderValue.String("second"),
+ }
+ )
+ with pytest.raises(ValueError, match="Distinct typed header keys"):
+ headers.to_scalar_dict()
+
+ def test_user_headers_construction_rejects_non_scalar_key(self):
+ with pytest.raises(
+ ValueError,
+ match=(
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue"
+ ),
+ ):
+ UserHeaders({object(): "value"})
+
+ def test_user_headers_construction_rejects_non_scalar_value(self):
+ with pytest.raises(
+ ValueError,
+ match=(
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue"
+ ),
+ ):
+ UserHeaders({"key": object()})
+
+ def test_user_headers_construction_rejects_invalid_keys_and_values(self):
+ with pytest.raises(
+ ValueError,
+ match=(
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue"
+ ),
+ ):
+ UserHeaders({object(): object()})
+
+ def test_user_headers_setitem_rejects_non_scalar_key(self):
+ headers = UserHeaders()
+ with pytest.raises(
+ ValueError,
+ match=(
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue"
+ ),
+ ):
+ headers[object()] = "value"
+
+ def test_user_headers_setitem_rejects_non_scalar_value(self):
+ headers = UserHeaders()
+ with pytest.raises(
+ ValueError,
+ match=(
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue"
+ ),
+ ):
+ headers["key"] = object()
+
+ def test_user_headers_to_scalar_dict_rejects_non_scalar_in_stored_data(
+ self,
+ ):
+ headers = UserHeaders()
+ dict.__setitem__(headers, object(), object())
+ with pytest.raises(
+ ValueError,
+ match=(
+ "User header must be str, bytes, bool, int, float, "
+ "HeaderKey, or HeaderValue"
+ ),
+ ):
+ headers.to_scalar_dict()
+
@pytest.mark.asyncio
@pytest.mark.parametrize(
"payload",
