slbotbm commented on code in PR #3613: URL: https://github.com/apache/iggy/pull/3613#discussion_r3593435776
########## 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) Review Comment: Even if init_system fails, this code will not know since init_system logs the error and returns normally. In produce_messages, the loop termination condition depends on successful sends, but if the send fails, `sent_batches` is not incremented after a failed send, a persistent configuration, authorization, or initialization failure makes the example run forever. It prints repeated errors instead of terminating with a useful failure. ########## examples/python/message-headers/common.py: ########## @@ -0,0 +1,247 @@ +# 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, Mapping +from dataclasses import dataclass +from enum import StrEnum +from typing import Any + +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(StrEnum): + 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 Review Comment: In this function, the generated `order_id`s are not coherent. The current code will effectively produce the following 6 events: ``` OrderConfirmed(order-0) OrderRejected(order-0) OrderCreated(order-3) OrderConfirmed(order-1) OrderRejected(order-1) OrderCreated(order-6) ``` ########## 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 are consumed identically, since plain headers are stored as typed headers on the wire. Review Comment: Changing this to > 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. would be better -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
