From c944ea6e9103bca6321517efba59ca9c121f47c7 Mon Sep 17 00:00:00 2001 From: Elioooon Date: Tue, 1 Sep 2026 14:27:35 +0800 Subject: [PATCH 1/4] feat(python): expose partition management --- foreign/python/apache_iggy.pyi | 40 ++++++++++++++++ foreign/python/src/client.rs | 66 ++++++++++++++++++++++++++ foreign/python/tests/test_partition.py | 41 ++++++++++++++++ 3 files changed, 147 insertions(+) create mode 100644 foreign/python/tests/test_partition.py diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 6c3715b54b..089370f0da 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -1191,6 +1191,46 @@ class IggyClient: Returns: An awaitable that resolves to `None` when the topic is purged. + Raises: + RuntimeError: If an identifier is invalid or the request fails. + """ + def create_partitions( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + partitions_count: builtins.int, + ) -> collections.abc.Awaitable[None]: + r""" + Create partitions for a topic. + + Args: + stream_id: Stream identifier as `str | int`. + topic_id: Topic identifier as `str | int`. + partitions_count: Number of partitions to create. + + Returns: + An awaitable that resolves to `None` when the partitions are created. + + Raises: + RuntimeError: If an identifier is invalid or the request fails. + """ + def delete_partitions( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + partitions_count: builtins.int, + ) -> collections.abc.Awaitable[None]: + r""" + Delete partitions from a topic. + + Args: + stream_id: Stream identifier as `str | int`. + topic_id: Topic identifier as `str | int`. + partitions_count: Number of partitions to delete. + + Returns: + An awaitable that resolves to `None` when the partitions are deleted. + Raises: RuntimeError: If an identifier is invalid or the request fails. """ diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index f669a87484..3fb391b1f2 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -777,6 +777,72 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. + /// + /// Raises: + /// RuntimeError: If an identifier is invalid or the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn create_partitions<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partitions_count: u32, + ) -> PyResult> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .create_partitions(&stream_id, &topic_id, partitions_count) + .await + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + }) + } + + /// Delete partitions from a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to delete. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are deleted. + /// + /// Raises: + /// RuntimeError: If an identifier is invalid or the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn delete_partitions<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partitions_count: u32, + ) -> PyResult> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .delete_partitions(&stream_id, &topic_id, partitions_count) + .await + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + }) + } + /// Create a consumer group for a stream and topic. /// /// Args: diff --git a/foreign/python/tests/test_partition.py b/foreign/python/tests/test_partition.py new file mode 100644 index 0000000000..89c16905c2 --- /dev/null +++ b/foreign/python/tests/test_partition.py @@ -0,0 +1,41 @@ +# 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 pytest + +from apache_iggy import IggyClient + + +@pytest.mark.asyncio +async def test_create_and_delete_partitions(iggy_client: IggyClient, unique_name): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=2 + ) + + await iggy_client.create_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 4 + + await iggy_client.delete_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 2 From 054dacc0e0e1fd45802622e84aa37d126cba7a85 Mon Sep 17 00:00:00 2001 From: Elioooon Date: Wed, 2 Sep 2026 10:33:12 +0800 Subject: [PATCH 2/4] test(python): clarify partition semantics --- foreign/python/apache_iggy.pyi | 6 ++++-- foreign/python/src/client.rs | 6 ++++-- foreign/python/tests/test_partition.py | 2 ++ 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 089370f0da..1797c62f0a 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -1212,7 +1212,8 @@ class IggyClient: An awaitable that resolves to `None` when the partitions are created. Raises: - RuntimeError: If an identifier is invalid or the request fails. + ValueError: If an identifier is invalid. + RuntimeError: If the request fails. """ def delete_partitions( self, @@ -1232,7 +1233,8 @@ class IggyClient: An awaitable that resolves to `None` when the partitions are deleted. Raises: - RuntimeError: If an identifier is invalid or the request fails. + ValueError: If an identifier is invalid. + RuntimeError: If the request fails. """ def create_consumer_group( self, diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 3fb391b1f2..c35b6dae03 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -788,7 +788,8 @@ impl IggyClient { /// An awaitable that resolves to `None` when the partitions are created. /// /// Raises: - /// RuntimeError: If an identifier is invalid or the request fails. + /// ValueError: If an identifier is invalid. + /// RuntimeError: If the request fails. #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] fn create_partitions<'a>( &self, @@ -821,7 +822,8 @@ impl IggyClient { /// An awaitable that resolves to `None` when the partitions are deleted. /// /// Raises: - /// RuntimeError: If an identifier is invalid or the request fails. + /// ValueError: If an identifier is invalid. + /// RuntimeError: If the request fails. #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] fn delete_partitions<'a>( &self, diff --git a/foreign/python/tests/test_partition.py b/foreign/python/tests/test_partition.py index 89c16905c2..f34fd7d426 100644 --- a/foreign/python/tests/test_partition.py +++ b/foreign/python/tests/test_partition.py @@ -34,8 +34,10 @@ async def test_create_and_delete_partitions(iggy_client: IggyClient, unique_name topic = await iggy_client.get_topic(stream_name, topic_name) assert topic is not None assert topic.partitions_count == 4 + assert [partition.id for partition in topic.partitions] == [0, 1, 2, 3] await iggy_client.delete_partitions(stream_name, topic_name, 2) topic = await iggy_client.get_topic(stream_name, topic_name) assert topic is not None assert topic.partitions_count == 2 + assert [partition.id for partition in topic.partitions] == [0, 1] From ec33ea74408c13ab246122b3b60daa99d035369f Mon Sep 17 00:00:00 2001 From: Elioooon Date: Wed, 2 Sep 2026 12:56:56 +0800 Subject: [PATCH 3/4] test(python): cover partition management errors --- foreign/python/tests/test_partition.py | 81 ++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/foreign/python/tests/test_partition.py b/foreign/python/tests/test_partition.py index f34fd7d426..72d44e0344 100644 --- a/foreign/python/tests/test_partition.py +++ b/foreign/python/tests/test_partition.py @@ -41,3 +41,84 @@ async def test_create_and_delete_partitions(iggy_client: IggyClient, unique_name assert topic is not None assert topic.partitions_count == 2 assert [partition.id for partition in topic.partitions] == [0, 1] + + +@pytest.mark.asyncio +async def test_partition_management_accepts_numeric_ids( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + await iggy_client.create_topic( + stream=stream.id, name=topic_name, partitions_count=2 + ) + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + + await iggy_client.create_partitions(stream.id, topic.id, 1) + await iggy_client.delete_partitions(stream.id, topic.id, 1) + + topic = await iggy_client.get_topic(stream.id, topic.id) + assert topic is not None + assert [partition.id for partition in topic.partitions] == [0, 1] + + +@pytest.mark.asyncio +async def test_partition_management_rejects_zero_count( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.create_partitions(stream_name, topic_name, 0) + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, topic_name, 0) + + +@pytest.mark.asyncio +async def test_delete_partitions_rejects_count_larger_than_topic( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, topic_name, 3) + + +@pytest.mark.asyncio +async def test_partition_management_rejects_missing_stream_or_topic( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + missing_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.create_partitions(missing_name, topic_name, 1) + with pytest.raises(RuntimeError): + await iggy_client.create_partitions(stream_name, missing_name, 1) + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(missing_name, topic_name, 1) + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, missing_name, 1) From 8080ce3db332253ea6cf252c4d8c9b4ebc1621d6 Mon Sep 17 00:00:00 2001 From: Elioooon Date: Thu, 3 Sep 2026 15:35:47 +0800 Subject: [PATCH 4/4] docs(python): clarify partition deletion semantics --- foreign/python/apache_iggy.pyi | 4 ++-- foreign/python/src/client.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 1797c62f0a..954299a3b1 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -1222,12 +1222,12 @@ class IggyClient: partitions_count: builtins.int, ) -> collections.abc.Awaitable[None]: r""" - Delete partitions from a topic. + Delete the last partitions from a topic, including all messages stored in them. Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - partitions_count: Number of partitions to delete. + partitions_count: Number of partitions to delete from the end of the topic. Returns: An awaitable that resolves to `None` when the partitions are deleted. diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index c35b6dae03..b2f2372dca 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -811,12 +811,12 @@ impl IggyClient { }) } - /// Delete partitions from a topic. + /// Delete the last partitions from a topic, including all messages stored in them. /// /// Args: /// stream_id: Stream identifier as `str | int`. /// topic_id: Topic identifier as `str | int`. - /// partitions_count: Number of partitions to delete. + /// partitions_count: Number of partitions to delete from the end of the topic. /// /// Returns: /// An awaitable that resolves to `None` when the partitions are deleted.