This is an automated email from the ASF dual-hosted git repository.
hubcio 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 c59961ce4 feat(python): add multi-tenant producer and consumer
examples (#4751)
c59961ce4 is described below
commit c59961ce4312c0723115bf83db896cf410103d4a
Author: yum <[email protected]>
AuthorDate: Thu Nov 5 22:26:32 2026 +0800
feat(python): add multi-tenant producer and consumer examples (#4751)
---
examples/python/README.md | 22 ++++-
examples/python/multi-tenant/consumer.py | 138 +++++++++++++++++++++++++++++++
examples/python/multi-tenant/producer.py | 121 +++++++++++++++++++++++++++
examples/python/multi-tenant/tenants.py | 123 +++++++++++++++++++++++++++
examples/python/pyproject.toml | 1 +
5 files changed, 404 insertions(+), 1 deletion(-)
diff --git a/examples/python/README.md b/examples/python/README.md
index ef57f1560..9c40c41ef 100644
--- a/examples/python/README.md
+++ b/examples/python/README.md
@@ -1,6 +1,6 @@
# Iggy Examples
-This directory contains Python SDK examples for connection configuration,
sending and polling messages, user headers, and TLS. To learn more about
building applications with Iggy, please refer to the [getting
started](https://iggy.apache.org/docs/introduction/getting-started) guide.
+This directory contains Python SDK examples for connection configuration,
sending and polling messages, user headers, multi-tenant permissions, and TLS.
To learn more about building applications with Iggy, please refer to the
[getting started](https://iggy.apache.org/docs/introduction/getting-started)
guide.
## Running Examples
@@ -183,6 +183,26 @@ python sink-data-producer/producer.py
IGGY_STREAM=test-stream IGGY_TOPIC=test-topic uv run
sink-data-producer/producer.py
```
+### Multi-Tenant
+
+Shows several tenants sharing one server, each limited to its own stream. The
producer connects as root and creates a
+stream and a user for each tenant, with permissions on that stream only. It
then connects as each tenant, checks that
+the other tenants' streams are refused, and sends messages to the `events`,
`logs` and `notifications` topics. The
+consumer creates a read-only user for one tenant, runs the same check, and
reads that tenant's messages.
+
+Both scripts read the tenant count from `TENANTS_COUNT` (default 3). The
consumer reads the tenant set by `TENANT_ID`
+(default 1). Run the producer first.
+
+```bash
+# Using uv
+uv run multi-tenant/producer.py
+uv run multi-tenant/consumer.py
+
+# Without using uv
+python multi-tenant/producer.py
+python multi-tenant/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/multi-tenant/consumer.py
b/examples/python/multi-tenant/consumer.py
new file mode 100644
index 000000000..5eac6a139
--- /dev/null
+++ b/examples/python/multi-tenant/consumer.py
@@ -0,0 +1,138 @@
+# 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
+from dataclasses import dataclass
+
+from apache_iggy import (
+ CommitPolicy,
+ Consumer,
+ IggyClient,
+ Permissions,
+ PollingStrategy,
+ PollOptions,
+ StreamPermissions,
+)
+from loguru import logger
+from tenants import (
+ PARTITION_ID,
+ TOPICS,
+ build_parser,
+ connect,
+ connect_as,
+ ensure_stream_access,
+ ensure_user,
+ read_positive_int,
+ stream_name,
+)
+
+CONSUMER_NAME = "multi-tenant"
+MESSAGES_PER_POLL = 100
+
+
+@dataclass(frozen=True, slots=True)
+class ArgNamespace:
+ connection_string: str
+ tenants_count: int
+ tenant_id: int
+
+
+def parse_args() -> ArgNamespace:
+ parser = build_parser(
+ "TENANTS_COUNT sets the number of tenants (default 3). "
+ "TENANT_ID sets the tenant that the consumer reads (default 1)."
+ )
+ args = parser.parse_args()
+ tenants_count = read_positive_int(parser, "TENANTS_COUNT", "3")
+ tenant_id = read_positive_int(parser, "TENANT_ID", "1")
+ if tenant_id > tenants_count:
+ parser.error(f"TENANT_ID {tenant_id} is above TENANTS_COUNT
{tenants_count}")
+ return ArgNamespace(args.connection_string, tenants_count, tenant_id)
+
+
+async def main() -> None:
+ args = parse_args()
+ logger.info(
+ f"Multi-tenant consumer has started, tenant: {args.tenant_id}, "
+ f"tenants: {args.tenants_count}"
+ )
+
+ logger.info("Connecting as root to create a read-only user for the tenant")
+ root = await connect(args.connection_string)
+ await create_consumer_user(root, args.tenant_id)
+ await root.shutdown()
+
+ client = await connect_as(args.connection_string,
consumer_username(args.tenant_id))
+ await ensure_stream_access(client, args.tenant_id, args.tenants_count)
+ await consume_messages(client, args.tenant_id)
+ await client.shutdown()
+
+ logger.info("Multi-tenant consumer has finished")
+
+
+async def create_consumer_user(root: IggyClient, tenant_id: int) -> None:
+ name = stream_name(tenant_id)
+ stream = await root.get_stream(name)
+ if stream is None:
+ raise RuntimeError(f"Stream {name} does not exist, run the producer
first")
+
+ # Reading the stream includes polling its topics, storing consumer offsets,
+ # and creating or deleting consumer groups in the stream. It does not allow
+ # sending messages or changing topics.
+ permissions = Permissions(streams={stream.id:
StreamPermissions(read_stream=True)})
+ await ensure_user(root, consumer_username(tenant_id), permissions)
+
+
+async def consume_messages(client: IggyClient, tenant_id: int) -> None:
+ stream = stream_name(tenant_id)
+ for topic in TOPICS:
+ consumed = 0
+ # The server stores the offset after each poll, so the next poll and
the
+ # next run continue after the last message read.
+ while True:
+ polled = await client.poll_messages(
+ stream,
+ topic,
+ partition_id=PARTITION_ID,
+ consumer=Consumer.Single(CONSUMER_NAME),
+ polling_strategy=PollingStrategy.Next(),
+ options=PollOptions(
+ count=MESSAGES_PER_POLL,
+ commit=CommitPolicy.OnRead(),
+ ),
+ )
+ if not polled.messages:
+ break
+ for message in polled.messages:
+ # The offset of this batch is already stored, so a payload that
+ # is not UTF-8 is logged with replacement characters instead of
+ # stopping the run and skipping the rest of the batch.
+ payload = message.payload().decode("utf-8", errors="replace")
+ logger.info(
+ f"Tenant {tenant_id} received {payload} from
{stream}/{topic} "
+ f"at offset {message.offset()}"
+ )
+ consumed += len(polled.messages)
+ logger.info(f"Tenant {tenant_id} consumed {consumed} messages from
{topic}")
+
+
+def consumer_username(tenant_id: int) -> str:
+ return f"tenant_{tenant_id}_consumer"
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/multi-tenant/producer.py
b/examples/python/multi-tenant/producer.py
new file mode 100644
index 000000000..bac3b6472
--- /dev/null
+++ b/examples/python/multi-tenant/producer.py
@@ -0,0 +1,121 @@
+# 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
+from dataclasses import dataclass
+
+from apache_iggy import IggyClient, Permissions, StreamPermissions
+from apache_iggy import SendMessage as Message
+from loguru import logger
+from tenants import (
+ PARTITION_ID,
+ TOPICS,
+ build_parser,
+ connect,
+ connect_as,
+ ensure_stream_access,
+ ensure_user,
+ read_positive_int,
+ stream_name,
+)
+
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+
+@dataclass(frozen=True, slots=True)
+class ArgNamespace:
+ connection_string: str
+ tenants_count: int
+
+
+def parse_args() -> ArgNamespace:
+ parser = build_parser("TENANTS_COUNT sets the number of tenants (default
3).")
+ args = parser.parse_args()
+ tenants_count = read_positive_int(parser, "TENANTS_COUNT", "3")
+ return ArgNamespace(args.connection_string, tenants_count)
+
+
+async def main() -> None:
+ args = parse_args()
+ tenant_ids = range(1, args.tenants_count + 1)
+ logger.info(f"Multi-tenant producer has started, tenants:
{args.tenants_count}")
+
+ logger.info("Connecting as root to create a stream and a user for each
tenant")
+ root = await connect(args.connection_string)
+ for tenant_id in tenant_ids:
+ await create_stream_and_user(root, tenant_id)
+ await root.shutdown()
+
+ for tenant_id in tenant_ids:
+ client = await connect_as(args.connection_string,
producer_username(tenant_id))
+ await ensure_stream_access(client, tenant_id, args.tenants_count)
+ await produce_messages(client, tenant_id)
+ await client.shutdown()
+
+ logger.info("Multi-tenant producer has finished")
+
+
+async def create_stream_and_user(root: IggyClient, tenant_id: int) -> None:
+ name = stream_name(tenant_id)
+ stream = await root.get_stream(name)
+ if stream is None:
+ stream = await root.create_stream(name=name)
+ logger.info(f"Created stream {name} with ID {stream.id}")
+ else:
+ logger.warning(f"Stream {name} already exists with ID {stream.id}")
+
+ # Managing topics includes reading them and sending messages. The user gets
+ # no global permissions and no other stream, so the rest of the server is
+ # closed to it.
+ permissions = Permissions(
+ streams={stream.id: StreamPermissions(read_stream=True,
manage_topics=True)}
+ )
+ await ensure_user(root, producer_username(tenant_id), permissions)
+
+
+async def produce_messages(client: IggyClient, tenant_id: int) -> None:
+ stream = stream_name(tenant_id)
+ for topic in TOPICS:
+ if await client.get_topic(stream, topic) is None:
+ await client.create_topic(stream=stream, name=topic)
+ logger.info(f"Tenant {tenant_id} created topic {topic} in
{stream}")
+
+ for batch in range(BATCHES_LIMIT):
+ first_id = batch * MESSAGES_PER_BATCH + 1
+ messages = [
+ Message(f"tenant-{tenant_id}-{topic}-{message_id}")
+ for message_id in range(first_id, first_id +
MESSAGES_PER_BATCH)
+ ]
+ await client.send_messages(
+ stream=stream,
+ topic=topic,
+ partitioning=PARTITION_ID,
+ messages=messages,
+ )
+ logger.info(
+ f"Tenant {tenant_id} sent {BATCHES_LIMIT * MESSAGES_PER_BATCH} "
+ f"messages to {stream}/{topic}"
+ )
+
+
+def producer_username(tenant_id: int) -> str:
+ return f"tenant_{tenant_id}_producer"
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/multi-tenant/tenants.py
b/examples/python/multi-tenant/tenants.py
new file mode 100644
index 000000000..a878cd67a
--- /dev/null
+++ b/examples/python/multi-tenant/tenants.py
@@ -0,0 +1,123 @@
+# 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 os
+
+from apache_iggy import IggyClient, IggyError, Permissions
+from loguru import logger
+
+TOPICS = ("events", "logs", "notifications")
+# One fixed password keeps the example short and lets it run again against the
+# same server. Real tenants would each get their own secret.
+TENANT_PASSWORD = "secret" # noqa: S105
+PARTITION_ID = 0
+UNAUTHORIZED = 41
+
+
+def build_parser(environment_help: str) -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ epilog=f"Environment variables: {environment_help}"
+ )
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string of the root user, e.g. "
+ "'iggy+tcp://iggy:[email protected]:8090'"
+ ),
+ default="iggy+tcp://iggy:[email protected]:8090",
+ nargs="?",
+ type=str,
+ )
+ return parser
+
+
+def read_positive_int(parser: argparse.ArgumentParser, name: str, default:
str) -> int:
+ value = os.environ.get(name, default)
+ if not value.isdecimal() or int(value) < 1:
+ parser.error(f"{name} must be a positive integer, got {value!r}")
+ return int(value)
+
+
+def stream_name(tenant_id: int) -> str:
+ return f"tenant_{tenant_id}_stream"
+
+
+async def connect(connection_string: str) -> IggyClient:
+ client = IggyClient.from_connection_string(connection_string)
+ await client.connect()
+ return client
+
+
+async def connect_as(root_connection_string: str, username: str) -> IggyClient:
+ logger.info(f"Connecting as {username}")
+ return await connect(tenant_connection_string(root_connection_string,
username))
+
+
+async def ensure_user(
+ root: IggyClient, username: str, permissions: Permissions
+) -> None:
+ if await root.get_user(username) is None:
+ await root.create_user(username, TENANT_PASSWORD,
permissions=permissions)
+ logger.info(f"Created user {username}")
+ return
+
+ # The stream may have been deleted and created again under a new ID since
+ # the last run, so the stored permissions are replaced, not trusted.
+ await root.update_permissions(username, permissions)
+ logger.warning(f"User {username} already exists, its permissions were
updated")
+
+
+async def ensure_stream_access(
+ client: IggyClient, tenant_id: int, tenants_count: int
+) -> None:
+ own_stream = stream_name(tenant_id)
+ if await client.get_stream(own_stream) is None:
+ raise RuntimeError(f"Tenant {tenant_id} cannot access stream
{own_stream}")
+ logger.info(f"Tenant {tenant_id} can access stream {own_stream}")
+
+ for other_id in range(1, tenants_count + 1):
+ if other_id == tenant_id:
+ continue
+ other_stream = stream_name(other_id)
+ try:
+ stream = await client.get_stream(other_stream)
+ except IggyError as error:
+ if error.code != UNAUTHORIZED:
+ raise
+ logger.info(
+ f"Tenant {tenant_id} has no access to stream {other_stream} "
+ "(unauthorized)"
+ )
+ continue
+ if stream is not None:
+ raise RuntimeError(
+ f"Tenant {tenant_id} must not have access to stream
{other_stream}"
+ )
+ logger.info(
+ f"Tenant {tenant_id} has no access to stream {other_stream} (not
visible)"
+ )
+
+
+def tenant_connection_string(root_connection_string: str, username: str) ->
str:
+ # Each tenant signs in as its own user, so only the credentials change. The
+ # string is split the way the SDK parser splits it, at "://" and at "@", so
+ # a root password with "/", "?" or "#" keeps the address intact. The SDK
+ # does not decode percent escapes, so the credentials go in as they are.
+ scheme, _, rest = root_connection_string.partition("://")
+ address = rest.rpartition("@")[2]
+ return f"{scheme}://{username}:{TENANT_PASSWORD}@{address}"
diff --git a/examples/python/pyproject.toml b/examples/python/pyproject.toml
index 0b0f46b17..f5b7dc549 100644
--- a/examples/python/pyproject.toml
+++ b/examples/python/pyproject.toml
@@ -58,4 +58,5 @@ search-path = [
"message-headers",
"message-headers/plain-headers",
"message-headers/typed-headers",
+ "multi-tenant",
]
