Skip to content

Commit 9803fda

Browse files
changhcchanghc
authored
feat(example): basic example for python sdk (#2118)
Created a basic example for the python SDK that sort of matches`examples/rust/basic`. It demonstrates how to configure clients using connection strings. Overall it's pretty similar to the `getting-started` example. This example is a bit different from `examples/rust/basic` because the Rust client builder struct and config types are not exposed through the python SDK as explained in #2110.
1 parent f3ff279 commit 9803fda

3 files changed

Lines changed: 242 additions & 0 deletions

File tree

‎examples/python/README.md‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,3 +43,14 @@ Perfect introduction for newcomers to Iggy:
4343
python getting-started/producer.py
4444
python getting-started/consumer.py
4545
```
46+
47+
### Basic Usage
48+
49+
Core functionality with detailed configuration options:
50+
51+
```bash
52+
python basic/producer.py <connection_string>
53+
python basic/consumer.py <connection_string>
54+
```
55+
56+
Demonstrates fundamental client connection, authentication, batch message sending, and polling with support for TCP/QUIC/HTTP protocols.

‎examples/python/basic/consumer.py‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
18+
import argparse
19+
import asyncio
20+
from collections import namedtuple
21+
22+
from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage
23+
from loguru import logger
24+
25+
STREAM_NAME = "sample-stream"
26+
TOPIC_NAME = "sample-topic"
27+
STREAM_ID = 1
28+
TOPIC_ID = 1
29+
PARTITION_ID = 1
30+
BATCHES_LIMIT = 5
31+
32+
ArgNamespace = namedtuple("ArgNamespace", ["connection_string"])
33+
34+
35+
def parse_args() -> argparse.Namespace:
36+
parser = argparse.ArgumentParser()
37+
parser.add_argument(
38+
"connection_string",
39+
help=(
40+
"Connection string for Iggy client, e.g. 'iggy+tcp://iggy:iggy@127.0.0.1:8090'"
41+
),
42+
default="iggy+tcp://iggy:iggy@127.0.0.1:8090",
43+
type=str,
44+
)
45+
return parser.parse_args()
46+
47+
48+
async def main():
49+
args: ArgNamespace = parse_args()
50+
client = IggyClient.from_connection_string(args.connection_string)
51+
logger.info("Connecting to Iggy")
52+
await client.connect()
53+
logger.info("Connected")
54+
await consume_messages(client)
55+
56+
57+
async def consume_messages(client: IggyClient):
58+
interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep
59+
logger.info(
60+
f"Messages will be consumed from stream: {STREAM_NAME}, topic: {TOPIC_NAME}, partition: {PARTITION_ID} with "
61+
f"interval {interval * 1000} ms."
62+
)
63+
offset = 0
64+
messages_per_batch = 10
65+
n_consumed_batches = 0
66+
while n_consumed_batches < BATCHES_LIMIT:
67+
try:
68+
logger.debug("Polling for messages...")
69+
polled_messages = await client.poll_messages(
70+
stream=STREAM_NAME,
71+
topic=TOPIC_NAME,
72+
partition_id=PARTITION_ID,
73+
polling_strategy=PollingStrategy.Next(),
74+
count=messages_per_batch,
75+
auto_commit=True,
76+
)
77+
if not polled_messages:
78+
logger.info("No messages found in current poll")
79+
await asyncio.sleep(interval)
80+
continue
81+
82+
offset += len(polled_messages)
83+
for message in polled_messages:
84+
handle_message(message)
85+
n_consumed_batches += 1
86+
await asyncio.sleep(interval)
87+
except Exception as error:
88+
logger.exception("Exception occurred while consuming messages: {}", error)
89+
break
90+
91+
logger.info(f"Consumed {n_consumed_batches} batches of messages, exiting.")
92+
93+
94+
def handle_message(message: ReceiveMessage):
95+
payload = message.payload().decode("utf-8")
96+
logger.info(
97+
f"Handling message at offset: {message.offset()} with payload: {payload}..."
98+
)
99+
100+
101+
if __name__ == "__main__":
102+
asyncio.run(main())

‎examples/python/basic/producer.py‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
18+
import argparse
19+
import asyncio
20+
from collections import namedtuple
21+
22+
from apache_iggy import IggyClient, StreamDetails, TopicDetails
23+
from apache_iggy import SendMessage as Message
24+
from loguru import logger
25+
26+
STREAM_NAME = "sample-stream"
27+
TOPIC_NAME = "sample-topic"
28+
STREAM_ID = 1
29+
TOPIC_ID = 1
30+
PARTITION_ID = 1
31+
BATCHES_LIMIT = 5
32+
33+
ArgNamespace = namedtuple("ArgNamespace", ["connection_string"])
34+
35+
36+
def parse_args() -> argparse.Namespace:
37+
parser = argparse.ArgumentParser()
38+
parser.add_argument(
39+
"connection_string",
40+
help=(
41+
"Connection string for Iggy client, e.g. 'iggy+tcp://iggy:iggy@127.0.0.1:8090'"
42+
),
43+
default="iggy+tcp://iggy:iggy@127.0.0.1:8090",
44+
type=str,
45+
)
46+
return parser.parse_args()
47+
48+
49+
async def main():
50+
args: ArgNamespace = parse_args()
51+
client = IggyClient.from_connection_string(args.connection_string)
52+
logger.info("Connecting to Iggy")
53+
await client.connect()
54+
logger.info("Connected")
55+
await init_system(client)
56+
await produce_messages(client)
57+
58+
59+
async def init_system(client: IggyClient):
60+
try:
61+
logger.info(f"Creating stream with name {STREAM_NAME}...")
62+
stream: StreamDetails = await client.get_stream(STREAM_NAME)
63+
if stream is None:
64+
await client.create_stream(name=STREAM_NAME, stream_id=STREAM_ID)
65+
logger.info("Stream was created successfully.")
66+
else:
67+
logger.warning(f"Stream {stream.name} already exists with ID {stream.id}")
68+
69+
except Exception as error:
70+
logger.error(f"Error creating stream: {error}")
71+
logger.exception(error)
72+
73+
try:
74+
logger.info(f"Creating topic {TOPIC_NAME} in stream {STREAM_NAME}")
75+
topic: TopicDetails = await client.get_topic(STREAM_NAME, TOPIC_NAME)
76+
if topic is None:
77+
await client.create_topic(
78+
stream=STREAM_NAME,
79+
partitions_count=1,
80+
name=TOPIC_NAME,
81+
replication_factor=1,
82+
)
83+
logger.info("Topic was created successfully.")
84+
else:
85+
logger.warning(f"Topic {topic.name} already exists with ID {topic.id}")
86+
except Exception as error:
87+
logger.error(f"Error creating topic {error}")
88+
logger.exception(error)
89+
90+
91+
async def produce_messages(client: IggyClient):
92+
interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep
93+
logger.info(
94+
f"Messages will be sent to stream: {STREAM_NAME}, topic: {TOPIC_NAME}, partition: {PARTITION_ID} with interval {interval * 1000} ms."
95+
)
96+
current_id = 0
97+
messages_per_batch = 10
98+
n_sent_batches = 0
99+
while n_sent_batches < BATCHES_LIMIT:
100+
messages = []
101+
for _ in range(messages_per_batch):
102+
current_id += 1
103+
payload = f"message-{current_id}"
104+
message = Message(payload)
105+
messages.append(message)
106+
logger.info(
107+
f"Attempting to send batch of {messages_per_batch} messages. Batch ID: {current_id // messages_per_batch}"
108+
)
109+
try:
110+
await client.send_messages(
111+
stream=STREAM_NAME,
112+
topic=TOPIC_NAME,
113+
partitioning=PARTITION_ID,
114+
messages=messages,
115+
)
116+
n_sent_batches += 1
117+
logger.info(
118+
f"Successfully sent batch of {messages_per_batch} messages. Batch ID: {current_id // messages_per_batch}"
119+
)
120+
except Exception as error:
121+
logger.error(f"Exception type: {type(error).__name__}, message: {error}")
122+
logger.exception(error)
123+
124+
await asyncio.sleep(interval)
125+
logger.info(f"Sent {n_sent_batches} batches of messages, exiting.")
126+
127+
128+
if __name__ == "__main__":
129+
asyncio.run(main())

0 commit comments

Comments
 (0)