From f598148baeb4b96156a6fc4661e96ae7f96dc272 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Mon, 6 Jul 2026 10:05:22 -0400
Subject: [PATCH 01/12] feat(python): expose remaining Topic/TopicDetails
 fields

---
 core/common/src/types/partition/mod.rs |   2 +-
 foreign/python/apache_iggy.pyi         | 169 +++++++++++++++++++++
 foreign/python/src/lib.rs              |   5 +-
 foreign/python/src/topic.rs            | 197 ++++++++++++++++++++++++-
 4 files changed, 369 insertions(+), 4 deletions(-)

diff --git a/core/common/src/types/partition/mod.rs b/core/common/src/types/partition/mod.rs
index 7b75151a98..fdd7161ab2 100644
--- a/core/common/src/types/partition/mod.rs
+++ b/core/common/src/types/partition/mod.rs
@@ -27,7 +27,7 @@ use serde::{Deserialize, Serialize};
 /// - `current_offset`: the current offset of the partition.
 /// - `size_bytes`: the size of the partition in bytes.
 /// - `messages_count`: the number of messages in the partition.
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Partition {
     /// Unique identifier of the partition.
     pub id: u32,
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 6a9ad26ba1..2788984a68 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -33,6 +33,9 @@ __all__ = [
     "ConsumerGroupMember",
     "IggyClient",
     "IggyConsumer",
+    "IggyExpiry",
+    "MaxTopicSize",
+    "Partition",
     "PollingStrategy",
     "ReceiveMessage",
     "SendMessage",
@@ -638,6 +641,117 @@ class IggyConsumer:
         Returns an awaitable that completes when shutdown is signaled or a PyRuntimeError on failure.
         """
 
+class IggyExpiry:
+    r"""
+    The expiry of the messages in a topic.
+    """
+    @typing.final
+    class ServerDefault(IggyExpiry):
+        r"""
+        Use the server's default message expiry.
+        """
+
+        __match_args__ = ()
+        def __new__(cls) -> IggyExpiry.ServerDefault: ...
+        def __len__(self) -> builtins.int: ...
+        def __getitem__(self, key: builtins.int, /) -> typing.Any: ...
+
+    @typing.final
+    class ExpireDuration(IggyExpiry):
+        r"""
+        Messages expire after this duration.
+        """
+
+        __match_args__ = ("duration",)
+        @property
+        def duration(self) -> datetime.timedelta: ...
+        def __new__(cls, duration: datetime.timedelta) -> IggyExpiry.ExpireDuration: ...
+
+    @typing.final
+    class NeverExpire(IggyExpiry):
+        r"""
+        Messages never expire.
+        """
+
+        __match_args__ = ()
+        def __new__(cls) -> IggyExpiry.NeverExpire: ...
+        def __len__(self) -> builtins.int: ...
+        def __getitem__(self, key: builtins.int, /) -> typing.Any: ...
+
+    ...
+
+class MaxTopicSize:
+    r"""
+    The maximum size of a topic.
+    """
+    @typing.final
+    class ServerDefault(MaxTopicSize):
+        r"""
+        Use the server's default max size.
+        """
+
+        __match_args__ = ()
+        def __new__(cls) -> MaxTopicSize.ServerDefault: ...
+        def __len__(self) -> builtins.int: ...
+        def __getitem__(self, key: builtins.int, /) -> typing.Any: ...
+
+    @typing.final
+    class Custom(MaxTopicSize):
+        r"""
+        The topic is limited to this many bytes.
+        """
+
+        __match_args__ = ("bytes",)
+        @property
+        def bytes(self) -> builtins.int: ...
+        def __new__(cls, bytes: builtins.int) -> MaxTopicSize.Custom: ...
+
+    @typing.final
+    class Unlimited(MaxTopicSize):
+        r"""
+        The topic has no maximum size.
+        """
+
+        __match_args__ = ()
+        def __new__(cls) -> MaxTopicSize.Unlimited: ...
+        def __len__(self) -> builtins.int: ...
+        def __getitem__(self, key: builtins.int, /) -> typing.Any: ...
+
+    ...
+
+@typing.final
+class Partition:
+    @property
+    def id(self) -> builtins.int:
+        r"""
+        The unique identifier (numeric) of the partition.
+        """
+    @property
+    def created_at(self) -> builtins.int:
+        r"""
+        The timestamp of the partition creation, in microseconds.
+        """
+    @property
+    def segments_count(self) -> builtins.int:
+        r"""
+        The number of segments in the partition.
+        """
+    @property
+    def current_offset(self) -> builtins.int:
+        r"""
+        The current offset of the partition.
+        """
+    @property
+    def size(self) -> builtins.int:
+        r"""
+        The size of the partition in bytes.
+        """
+    @property
+    def messages_count(self) -> builtins.int:
+        r"""
+        The number of messages in the partition.
+        """
+
 class PollingStrategy:
     @typing.final
     class Offset(PollingStrategy):
@@ -758,6 +872,36 @@ class Topic:
         r"""
         The total number of partitions in the topic.
         """
+    @property
+    def created_at(self) -> builtins.int:
+        r"""
+        The timestamp when the topic was created, in microseconds.
+        """
+    @property
+    def size(self) -> builtins.int:
+        r"""
+        The total size of the topic in bytes.
+        """
+    @property
+    def message_expiry(self) -> IggyExpiry:
+        r"""
+        The expiry of the messages in the topic.
+        """
+    @property
+    def compression_algorithm(self) -> builtins.str:
+        r"""
+        Compression algorithm for the topic.
+        """
+    @property
+    def max_topic_size(self) -> MaxTopicSize:
+        r"""
+        The maximum size of the topic.
+        """
+    @property
+    def replication_factor(self) -> builtins.int:
+        r"""
+        Replication factor for the topic.
+        """
 
 @typing.final
 class TopicDetails:
@@ -782,12 +926,37 @@ class TopicDetails:
         The total number of partitions in the topic.
         """
     @property
+    def created_at(self) -> builtins.int:
+        r"""
+        The timestamp when the topic was created, in microseconds.
+        """
+    @property
+    def size(self) -> builtins.int:
+        r"""
+        The total size of the topic in bytes.
+        """
+    @property
+    def message_expiry(self) -> IggyExpiry:
+        r"""
+        The expiry of the messages in the topic.
+        """
+    @property
     def compression_algorithm(self) -> builtins.str:
         r"""
         Compression algorithm for the topic.
         """
     @property
+    def max_topic_size(self) -> MaxTopicSize:
+        r"""
+        The maximum size of the topic.
+        """
+    @property
     def replication_factor(self) -> builtins.int:
         r"""
         Replication factor for the topic.
         """
+    @property
+    def partitions(self) -> builtins.list[Partition]:
+        r"""
+        The collection of partitions in the topic.
+        """
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 8c78456062..22c7f3d132 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -32,7 +32,7 @@ use pyo3::prelude::*;
 use receive_message::{PollingStrategy, ReceiveMessage};
 use send_message::SendMessage;
 use stream::StreamDetails;
-use topic::{Topic, TopicDetails};
+use topic::{IggyExpiry, MaxTopicSize, Partition, Topic, TopicDetails};
 
 /// A Python module implemented in Rust.
 #[pymodule]
@@ -43,6 +43,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<StreamDetails>()?;
     m.add_class::<Topic>()?;
     m.add_class::<TopicDetails>()?;
+    m.add_class::<IggyExpiry>()?;
+    m.add_class::<MaxTopicSize>()?;
+    m.add_class::<Partition>()?;
     m.add_class::<ConsumerGroup>()?;
     m.add_class::<ConsumerGroupDetails>()?;
     m.add_class::<ConsumerGroupMember>()?;
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index 0c9028ef4c..7b5066a0b2 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -15,9 +15,79 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use iggy::prelude::{Topic as RustTopic, TopicDetails as RustTopicDetails};
+use std::time::Duration;
+
+use iggy::prelude::{
+    IggyExpiry as RustIggyExpiry, MaxTopicSize as RustMaxTopicSize, Partition as RustPartition,
+    Topic as RustTopic, TopicDetails as RustTopicDetails,
+};
 use pyo3::prelude::*;
-use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
+use pyo3::types::PyDelta;
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum, gen_stub_pymethods};
+
+/// The expiry of the messages in a topic.
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+pub enum IggyExpiry {
+    /// Use the server's default message expiry.
+    ServerDefault(),
+    /// Messages expire after this duration.
+    ExpireDuration { duration: Py<PyDelta> },
+    /// Messages never expire.
+    NeverExpire(),
+}
+
+impl From<RustIggyExpiry> for IggyExpiry {
+    fn from(expiry: RustIggyExpiry) -> Self {
+        match expiry {
+            RustIggyExpiry::ServerDefault => IggyExpiry::ServerDefault(),
+            RustIggyExpiry::ExpireDuration(duration) => IggyExpiry::ExpireDuration {
+                duration: iggy_duration_to_py_delta(duration.get_duration()),
+            },
+            RustIggyExpiry::NeverExpire => IggyExpiry::NeverExpire(),
+        }
+    }
+}
+
+fn iggy_duration_to_py_delta(duration: Duration) -> Py<PyDelta> {
+    let days = duration.as_secs() / 86_400;
+    let secs_of_day = duration.as_secs() % 86_400;
+    Python::attach(|py| {
+        PyDelta::new(
+            py,
+            days as i32,
+            secs_of_day as i32,
+            duration.subsec_micros() as i32,
+            true,
+        )
+        .expect("topic message expiry duration fits within timedelta bounds")
+        .unbind()
+    })
+}
+
+/// The maximum size of a topic.
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+pub enum MaxTopicSize {
+    /// Use the server's default max size.
+    ServerDefault(),
+    /// The topic is limited to this many bytes.
+    Custom { bytes: u64 },
+    /// The topic has no maximum size.
+    Unlimited(),
+}
+
+impl From<RustMaxTopicSize> for MaxTopicSize {
+    fn from(max_size: RustMaxTopicSize) -> Self {
+        match max_size {
+            RustMaxTopicSize::ServerDefault => MaxTopicSize::ServerDefault(),
+            RustMaxTopicSize::Custom(size) => MaxTopicSize::Custom {
+                bytes: size.as_bytes_u64(),
+            },
+            RustMaxTopicSize::Unlimited => MaxTopicSize::Unlimited(),
+        }
+    }
+}
 
 #[gen_stub_pyclass]
 #[pyclass]
@@ -57,6 +127,42 @@ impl Topic {
     pub fn partitions_count(&self) -> u32 {
         self.inner.partitions_count
     }
+
+    /// The timestamp when the topic was created, in microseconds.
+    #[getter]
+    pub fn created_at(&self) -> u64 {
+        self.inner.created_at.as_micros()
+    }
+
+    /// The total size of the topic in bytes.
+    #[getter]
+    pub fn size(&self) -> u64 {
+        self.inner.size.as_bytes_u64()
+    }
+
+    /// The expiry of the messages in the topic.
+    #[getter]
+    pub fn message_expiry(&self) -> IggyExpiry {
+        self.inner.message_expiry.into()
+    }
+
+    /// Compression algorithm for the topic.
+    #[getter]
+    pub fn compression_algorithm(&self) -> String {
+        self.inner.compression_algorithm.to_string()
+    }
+
+    /// The maximum size of the topic.
+    #[getter]
+    pub fn max_topic_size(&self) -> MaxTopicSize {
+        self.inner.max_topic_size.into()
+    }
+
+    /// Replication factor for the topic.
+    #[getter]
+    pub fn replication_factor(&self) -> u8 {
+        self.inner.replication_factor
+    }
 }
 
 #[gen_stub_pyclass]
@@ -100,15 +206,102 @@ impl TopicDetails {
         self.inner.partitions_count
     }
 
+    /// The timestamp when the topic was created, in microseconds.
+    #[getter]
+    pub fn created_at(&self) -> u64 {
+        self.inner.created_at.as_micros()
+    }
+
+    /// The total size of the topic in bytes.
+    #[getter]
+    pub fn size(&self) -> u64 {
+        self.inner.size.as_bytes_u64()
+    }
+
+    /// The expiry of the messages in the topic.
+    #[getter]
+    pub fn message_expiry(&self) -> IggyExpiry {
+        self.inner.message_expiry.into()
+    }
+
     /// Compression algorithm for the topic.
     #[getter]
     pub fn compression_algorithm(&self) -> String {
         self.inner.compression_algorithm.to_string()
     }
 
+    /// The maximum size of the topic.
+    #[getter]
+    pub fn max_topic_size(&self) -> MaxTopicSize {
+        self.inner.max_topic_size.into()
+    }
+
     /// Replication factor for the topic.
     #[getter]
     pub fn replication_factor(&self) -> u8 {
         self.inner.replication_factor
     }
+
+    /// The collection of partitions in the topic.
+    #[getter]
+    pub fn partitions(&self) -> Vec<Partition> {
+        self.inner
+            .partitions
+            .iter()
+            .cloned()
+            .map(Partition::from)
+            .collect()
+    }
+}
+
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct Partition {
+    pub(crate) inner: RustPartition,
+}
+
+impl From<RustPartition> for Partition {
+    fn from(partition: RustPartition) -> Self {
+        Self { inner: partition }
+    }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Partition {
+    /// The unique identifier (numeric) of the partition.
+    #[getter]
+    pub fn id(&self) -> u32 {
+        self.inner.id
+    }
+
+    /// The timestamp of the partition creation, in microseconds.
+    #[getter]
+    pub fn created_at(&self) -> u64 {
+        self.inner.created_at.as_micros()
+    }
+
+    /// The number of segments in the partition.
+    #[getter]
+    pub fn segments_count(&self) -> u32 {
+        self.inner.segments_count
+    }
+
+    /// The current offset of the partition.
+    #[getter]
+    pub fn current_offset(&self) -> u64 {
+        self.inner.current_offset
+    }
+
+    /// The size of the partition in bytes.
+    #[getter]
+    pub fn size(&self) -> u64 {
+        self.inner.size.as_bytes_u64()
+    }
+
+    /// The number of messages in the partition.
+    #[getter]
+    pub fn messages_count(&self) -> u64 {
+        self.inner.messages_count
+    }
 }

From 159c023d001c80803dc923a16603557eb8154193 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Mon, 6 Jul 2026 13:48:31 -0400
Subject: [PATCH 02/12] test(python): assert new Topic/TopicDetails fields in
 topic tests

---
 foreign/python/tests/test_topic.py | 93 +++++++++++++++++++++++++-----
 1 file changed, 80 insertions(+), 13 deletions(-)

diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index f3acc53954..4775d08900 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -19,7 +19,7 @@
 
 import pytest
 
-from apache_iggy import IggyClient, SendMessage
+from apache_iggy import IggyClient, IggyExpiry, MaxTopicSize, SendMessage
 
 from .utils import get_server_config, wait_for_ping, wait_for_server
 
@@ -70,6 +70,10 @@ async def test_create_and_get_topic(
         assert topic is not None
         assert topic.name == topic_name
         assert topic.partitions_count == 2
+        assert topic.created_at > 0
+        assert topic.size == 0
+        assert len(topic.partitions) == 2
+        assert all(partition.messages_count == 0 for partition in topic.partitions)
 
         stream = await iggy_client.get_stream(stream_name)
         assert stream is not None
@@ -254,6 +258,13 @@ async def test_create_topic_with_message_expiry(
         topic = await iggy_client.get_topic(stream_name, topic_name)
         assert topic is not None
         assert topic.name == topic_name
+        if message_expiry == timedelta(0):
+            # A zero duration is the server-default sentinel; the server
+            # resolves it to the configured default, which is "never expire".
+            assert isinstance(topic.message_expiry, IggyExpiry.NeverExpire)
+        else:
+            assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
+            assert topic.message_expiry.duration == message_expiry
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("invalid_message_expiry", [1, "1s", object()])
@@ -276,15 +287,19 @@ async def test_create_topic_invalid_message_expiry(
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        "max_topic_size",
+        ("max_topic_size", "expected_kind"),
         [
-            0,  # value for server default max topic size
-            2**64 - 1,
-            2_000_000_000,
+            (0, "unlimited"),  # server-default sentinel; create_topic resolves it
+            (2**64 - 1, "unlimited"),  # explicit unlimited sentinel
+            (2_000_000_000, "custom"),
         ],
     )
     async def test_create_topic_with_valid_max_topic_size(
-        self, iggy_client: IggyClient, unique_name, max_topic_size: int
+        self,
+        iggy_client: IggyClient,
+        unique_name,
+        max_topic_size: int,
+        expected_kind: str,
     ):
         """Test create_topic accepts supported maximum topic size values."""
         stream_name = unique_name()
@@ -301,6 +316,11 @@ async def test_create_topic_with_valid_max_topic_size(
         topic = await iggy_client.get_topic(stream_name, topic_name)
         assert topic is not None
         assert topic.name == topic_name
+        if expected_kind == "unlimited":
+            assert isinstance(topic.max_topic_size, MaxTopicSize.Unlimited)
+        else:
+            assert isinstance(topic.max_topic_size, MaxTopicSize.Custom)
+            assert topic.max_topic_size.bytes == max_topic_size
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
@@ -540,6 +560,28 @@ async def test_get_topic_by_name_and_id(self, iggy_client: IggyClient, unique_na
         assert topic_by_id.id == topic_by_name.id
         assert topic_by_id.name == topic_by_name.name
 
+    @pytest.mark.asyncio
+    async def test_get_topic_partitions(self, iggy_client: IggyClient, unique_name):
+        """Test TopicDetails.partitions returns one Partition per partition."""
+        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=3
+        )
+
+        topic = await iggy_client.get_topic(stream_name, topic_name)
+        assert topic is not None
+        assert len(topic.partitions) == 3
+        assert [partition.id for partition in topic.partitions] == [0, 1, 2]
+        for partition in topic.partitions:
+            assert partition.created_at > 0
+            assert partition.segments_count == 1
+            assert partition.current_offset == 0
+            assert partition.size == 0
+            assert partition.messages_count == 0
+
     @pytest.mark.asyncio
     async def test_get_nonexistent_topic(self, iggy_client: IggyClient, unique_name):
         """Test getting a non-existent topic by name or numeric id."""
@@ -948,13 +990,27 @@ async def test_update_topic_with_message_expiry(
         topic = await iggy_client.get_topic(stream_name, topic_name)
         assert topic is not None
         assert topic.name == topic_name
-        # TODO: assert topic.message_expiry once TopicDetails exposes that
-        # getter (tracked for a follow-up PR).
+        assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
+        assert topic.message_expiry.duration == timedelta(minutes=10)
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("max_topic_size", [0, 2_000_000_000, 2**64 - 1])
+    @pytest.mark.parametrize(
+        ("max_topic_size", "expected_kind"),
+        [
+            # Unlike create_topic, update_topic does not resolve the
+            # server-default sentinel (0) to a concrete value, so it round
+            # trips as-is instead of becoming "unlimited".
+            (0, "server_default"),
+            (2_000_000_000, "custom"),
+            (2**64 - 1, "unlimited"),
+        ],
+    )
     async def test_update_topic_with_valid_max_topic_size(
-        self, iggy_client: IggyClient, unique_name, max_topic_size: int
+        self,
+        iggy_client: IggyClient,
+        unique_name,
+        max_topic_size: int,
+        expected_kind: str,
     ):
         """Test update_topic accepts supported maximum topic size values."""
         stream_name = unique_name()
@@ -975,8 +1031,13 @@ async def test_update_topic_with_valid_max_topic_size(
         topic = await iggy_client.get_topic(stream_name, topic_name)
         assert topic is not None
         assert topic.name == topic_name
-        # TODO: assert topic.message_expiry and topic.max_topic_size once
-        # TopicDetails exposes those getters (tracked for a follow-up PR).
+        if expected_kind == "server_default":
+            assert isinstance(topic.max_topic_size, MaxTopicSize.ServerDefault)
+        elif expected_kind == "unlimited":
+            assert isinstance(topic.max_topic_size, MaxTopicSize.Unlimited)
+        else:
+            assert isinstance(topic.max_topic_size, MaxTopicSize.Custom)
+            assert topic.max_topic_size.bytes == max_topic_size
 
     @pytest.mark.asyncio
     async def test_update_topic_applies_repeated_updates(
@@ -1213,12 +1274,18 @@ async def test_purge_topic_clears_messages_but_keeps_topic(
         after = await iggy_client.get_topic(stream_name, topic_name)
         assert after is not None
         assert after.messages_count == 0
-        # Purging clears messages only; every other field is left unchanged.
+        assert after.size == 0
+        # Purging clears messages and size only; topic config is unchanged.
         assert after.id == before.id
         assert after.name == before.name
+        assert after.created_at == before.created_at
         assert after.partitions_count == before.partitions_count
         assert after.compression_algorithm == before.compression_algorithm
         assert after.replication_factor == before.replication_factor
+        assert isinstance(before.message_expiry, IggyExpiry.NeverExpire)
+        assert isinstance(after.message_expiry, IggyExpiry.NeverExpire)
+        assert isinstance(before.max_topic_size, MaxTopicSize.Unlimited)
+        assert isinstance(after.max_topic_size, MaxTopicSize.Unlimited)
 
     @pytest.mark.asyncio
     async def test_purge_empty_topic_succeeds(

From 375860784f9d54634ee0a79846890a18f33b7eac Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Fri, 10 Jul 2026 19:16:50 -0400
Subject: [PATCH 03/12] feat(python): accept IggyExpiry/MaxTopicSize enums in
 create_topic and update_topic

---
 foreign/python/apache_iggy.pyi     | 12 ++---
 foreign/python/src/client.rs       | 41 ++++++++---------
 foreign/python/src/topic.rs        | 28 +++++++++++-
 foreign/python/tests/test_topic.py | 70 ++++++++++++++++--------------
 4 files changed, 91 insertions(+), 60 deletions(-)

diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 2788984a68..fbf566cef6 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -362,8 +362,8 @@ class IggyClient:
         partitions_count: builtins.int,
         compression_algorithm: builtins.str | None = None,
         replication_factor: builtins.int | None = None,
-        message_expiry: datetime.timedelta | None = None,
-        max_topic_size: builtins.int | None = None,
+        message_expiry: IggyExpiry | None = None,
+        max_topic_size: MaxTopicSize | None = None,
     ) -> collections.abc.Awaitable[None]:
         r"""
         Creates a new topic with the given parameters.
@@ -400,8 +400,8 @@ class IggyClient:
         name: builtins.str,
         compression_algorithm: builtins.str | None = None,
         replication_factor: builtins.int | None = None,
-        message_expiry: datetime.timedelta | None = None,
-        max_topic_size: builtins.int | None = None,
+        message_expiry: IggyExpiry | None = None,
+        max_topic_size: MaxTopicSize | None = None,
     ) -> collections.abc.Awaitable[None]:
         r"""
         Update an existing topic.
@@ -415,8 +415,8 @@ class IggyClient:
             name: New topic name as `str`.
             compression_algorithm: Compression algorithm as `str | None`.
             replication_factor: Replication factor as `int | None`.
-            message_expiry: Message expiry as `datetime.timedelta | None`.
-            max_topic_size: Maximum topic size in bytes as `int | None`.
+            message_expiry: Message expiry as `IggyExpiry | None`.
+            max_topic_size: Maximum topic size as `MaxTopicSize | None`.
 
         Returns:
             An awaitable that resolves to `None` when the topic is updated.
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index b604f62c6d..bfc41f40c8 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -16,7 +16,8 @@
 // under the License.
 
 use iggy::prelude::{
-    Consumer as RustConsumer, IggyClient as RustIggyClient, IggyMessage as RustMessage,
+    Consumer as RustConsumer, IggyClient as RustIggyClient, IggyExpiry as RustIggyExpiry,
+    IggyMessage as RustMessage, MaxTopicSize as RustMaxTopicSize,
     PollingStrategy as RustPollingStrategy, *,
 };
 use pyo3::PyRef;
@@ -36,7 +37,7 @@ use crate::identifier::PyIdentifier;
 use crate::receive_message::{PollingStrategy, ReceiveMessage};
 use crate::send_message::SendMessage;
 use crate::stream::StreamDetails;
-use crate::topic::{Topic, TopicDetails};
+use crate::topic::{IggyExpiry, MaxTopicSize, Topic, TopicDetails};
 use tokio::sync::Mutex;
 
 /// A Python class representing the Iggy client.
@@ -187,9 +188,12 @@ impl IggyClient {
         #[gen_stub(override_type(type_repr = "builtins.int | None"))] replication_factor: Option<
             u8,
         >,
-        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
-        message_expiry: Option<Py<PyDelta>>,
-        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_topic_size: Option<u64>,
+        #[gen_stub(override_type(type_repr = "IggyExpiry | None"))] message_expiry: Option<
+            &IggyExpiry,
+        >,
+        #[gen_stub(override_type(type_repr = "MaxTopicSize | None"))] max_topic_size: Option<
+            &MaxTopicSize,
+        >,
     ) -> PyResult<Bound<'a, PyAny>> {
         let compression_algorithm = match compression_algorithm {
             Some(algo) => CompressionAlgorithm::from_str(&algo)
@@ -197,12 +201,9 @@ impl IggyClient {
             None => CompressionAlgorithm::default(),
         };
 
-        let expiry = match message_expiry {
-            Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)),
-            None => IggyExpiry::ServerDefault,
-        };
+        let expiry = message_expiry.map_or(RustIggyExpiry::ServerDefault, RustIggyExpiry::from);
 
-        let max_size = max_topic_size.map_or(MaxTopicSize::ServerDefault, MaxTopicSize::from);
+        let max_size = max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
 
         let stream = Identifier::try_from(stream)?;
         let inner = self.inner.clone();
@@ -285,8 +286,8 @@ impl IggyClient {
     ///     name: New topic name as `str`.
     ///     compression_algorithm: Compression algorithm as `str | None`.
     ///     replication_factor: Replication factor as `int | None`.
-    ///     message_expiry: Message expiry as `datetime.timedelta | None`.
-    ///     max_topic_size: Maximum topic size in bytes as `int | None`.
+    ///     message_expiry: Message expiry as `IggyExpiry | None`.
+    ///     max_topic_size: Maximum topic size as `MaxTopicSize | None`.
     ///
     /// Returns:
     ///     An awaitable that resolves to `None` when the topic is updated.
@@ -310,9 +311,12 @@ impl IggyClient {
         #[gen_stub(override_type(type_repr = "builtins.int | None"))] replication_factor: Option<
             u8,
         >,
-        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
-        message_expiry: Option<Py<PyDelta>>,
-        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_topic_size: Option<u64>,
+        #[gen_stub(override_type(type_repr = "IggyExpiry | None"))] message_expiry: Option<
+            &IggyExpiry,
+        >,
+        #[gen_stub(override_type(type_repr = "MaxTopicSize | None"))] max_topic_size: Option<
+            &MaxTopicSize,
+        >,
     ) -> PyResult<Bound<'a, PyAny>> {
         let compression_algorithm = match compression_algorithm {
             Some(algo) => CompressionAlgorithm::from_str(&algo)
@@ -320,12 +324,9 @@ impl IggyClient {
             None => CompressionAlgorithm::default(),
         };
 
-        let expiry = match message_expiry {
-            Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)),
-            None => IggyExpiry::ServerDefault,
-        };
+        let expiry = message_expiry.map_or(RustIggyExpiry::ServerDefault, RustIggyExpiry::from);
 
-        let max_size = max_topic_size.map_or(MaxTopicSize::ServerDefault, MaxTopicSize::from);
+        let max_size = max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
 
         let stream_id = Identifier::try_from(stream_id)?;
         let topic_id = Identifier::try_from(topic_id)?;
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index 7b5066a0b2..2c042087a1 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -18,13 +18,15 @@
 use std::time::Duration;
 
 use iggy::prelude::{
-    IggyExpiry as RustIggyExpiry, MaxTopicSize as RustMaxTopicSize, Partition as RustPartition,
-    Topic as RustTopic, TopicDetails as RustTopicDetails,
+    IggyByteSize, IggyExpiry as RustIggyExpiry, MaxTopicSize as RustMaxTopicSize,
+    Partition as RustPartition, Topic as RustTopic, TopicDetails as RustTopicDetails,
 };
 use pyo3::prelude::*;
 use pyo3::types::PyDelta;
 use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum, gen_stub_pymethods};
 
+use crate::consumer::py_delta_to_iggy_duration;
+
 /// The expiry of the messages in a topic.
 #[gen_stub_pyclass_complex_enum]
 #[pyclass]
@@ -49,6 +51,18 @@ impl From<RustIggyExpiry> for IggyExpiry {
     }
 }
 
+impl From<&IggyExpiry> for RustIggyExpiry {
+    fn from(expiry: &IggyExpiry) -> Self {
+        match expiry {
+            IggyExpiry::ServerDefault() => RustIggyExpiry::ServerDefault,
+            IggyExpiry::ExpireDuration { duration } => {
+                RustIggyExpiry::ExpireDuration(py_delta_to_iggy_duration(duration))
+            }
+            IggyExpiry::NeverExpire() => RustIggyExpiry::NeverExpire,
+        }
+    }
+}
+
 fn iggy_duration_to_py_delta(duration: Duration) -> Py<PyDelta> {
     let days = duration.as_secs() / 86_400;
     let secs_of_day = duration.as_secs() % 86_400;
@@ -89,6 +103,16 @@ impl From<RustMaxTopicSize> for MaxTopicSize {
     }
 }
 
+impl From<&MaxTopicSize> for RustMaxTopicSize {
+    fn from(max_size: &MaxTopicSize) -> Self {
+        match max_size {
+            MaxTopicSize::ServerDefault() => RustMaxTopicSize::ServerDefault,
+            MaxTopicSize::Custom { bytes } => RustMaxTopicSize::Custom(IggyByteSize::from(*bytes)),
+            MaxTopicSize::Unlimited() => RustMaxTopicSize::Unlimited,
+        }
+    }
+}
+
 #[gen_stub_pyclass]
 #[pyclass]
 pub struct Topic {
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 4775d08900..47658e8785 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -233,15 +233,18 @@ async def test_create_topic_invalid_compression_algorithm(
     @pytest.mark.parametrize(
         "message_expiry",
         [
-            timedelta(0),  # value for server default message expiry
-            timedelta(microseconds=1),
-            timedelta(seconds=1),
-            timedelta(minutes=10),
-            timedelta(days=1, seconds=2, microseconds=3),
+            IggyExpiry.ExpireDuration(timedelta(0)),  # server default sentinel
+            IggyExpiry.ExpireDuration(timedelta(microseconds=1)),
+            IggyExpiry.ExpireDuration(timedelta(seconds=1)),
+            IggyExpiry.ExpireDuration(timedelta(minutes=10)),
+            IggyExpiry.ExpireDuration(timedelta(days=1, seconds=2, microseconds=3)),
         ],
     )
     async def test_create_topic_with_message_expiry(
-        self, iggy_client: IggyClient, unique_name, message_expiry: timedelta
+        self,
+        iggy_client: IggyClient,
+        unique_name,
+        message_expiry: IggyExpiry.ExpireDuration,
     ):
         """Test create_topic accepts an explicit message expiry."""
         stream_name = unique_name()
@@ -258,20 +261,22 @@ async def test_create_topic_with_message_expiry(
         topic = await iggy_client.get_topic(stream_name, topic_name)
         assert topic is not None
         assert topic.name == topic_name
-        if message_expiry == timedelta(0):
+        if message_expiry.duration == timedelta(0):
             # A zero duration is the server-default sentinel; the server
             # resolves it to the configured default, which is "never expire".
             assert isinstance(topic.message_expiry, IggyExpiry.NeverExpire)
         else:
             assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
-            assert topic.message_expiry.duration == message_expiry
+            assert topic.message_expiry.duration == message_expiry.duration
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("invalid_message_expiry", [1, "1s", object()])
+    @pytest.mark.parametrize(
+        "invalid_message_expiry", [1, "1s", object(), timedelta(seconds=1)]
+    )
     async def test_create_topic_invalid_message_expiry(
         self, iggy_client: IggyClient, unique_name, invalid_message_expiry
     ):
-        """Test create_topic rejects message_expiry values that are not timedeltas."""
+        """Test create_topic rejects non-IggyExpiry message_expiry values."""
         stream_name = unique_name()
         topic_name = unique_name()
 
@@ -289,16 +294,16 @@ async def test_create_topic_invalid_message_expiry(
     @pytest.mark.parametrize(
         ("max_topic_size", "expected_kind"),
         [
-            (0, "unlimited"),  # server-default sentinel; create_topic resolves it
-            (2**64 - 1, "unlimited"),  # explicit unlimited sentinel
-            (2_000_000_000, "custom"),
+            (MaxTopicSize.ServerDefault(), "unlimited"),  # resolved by create_topic
+            (MaxTopicSize.Unlimited(), "unlimited"),
+            (MaxTopicSize.Custom(2_000_000_000), "custom"),
         ],
     )
     async def test_create_topic_with_valid_max_topic_size(
         self,
         iggy_client: IggyClient,
         unique_name,
-        max_topic_size: int,
+        max_topic_size: MaxTopicSize,
         expected_kind: str,
     ):
         """Test create_topic accepts supported maximum topic size values."""
@@ -320,11 +325,12 @@ async def test_create_topic_with_valid_max_topic_size(
             assert isinstance(topic.max_topic_size, MaxTopicSize.Unlimited)
         else:
             assert isinstance(topic.max_topic_size, MaxTopicSize.Custom)
-            assert topic.max_topic_size.bytes == max_topic_size
+            assert isinstance(max_topic_size, MaxTopicSize.Custom)
+            assert topic.max_topic_size.bytes == max_topic_size.bytes
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        ("max_topic_size", "expected_exception"),
+        ("max_topic_size_bytes", "expected_exception"),
         [
             (4563, RuntimeError),
             (-1, OverflowError),
@@ -335,7 +341,7 @@ async def test_create_topic_invalid_max_topic_size(
         self,
         iggy_client: IggyClient,
         unique_name,
-        max_topic_size,
+        max_topic_size_bytes,
         expected_exception,
     ):
         """Test create_topic rejects invalid maximum topic size values."""
@@ -349,7 +355,7 @@ async def test_create_topic_invalid_max_topic_size(
                 stream=stream_name,
                 name=topic_name,
                 partitions_count=1,
-                max_topic_size=max_topic_size,
+                max_topic_size=MaxTopicSize.Custom(max_topic_size_bytes),
             )
 
     @pytest.mark.asyncio
@@ -854,11 +860,13 @@ async def test_update_topic_invalid_compression_algorithm(
             )
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("invalid_message_expiry", [1, "1s", object()])
+    @pytest.mark.parametrize(
+        "invalid_message_expiry", [1, "1s", object(), timedelta(seconds=1)]
+    )
     async def test_update_topic_invalid_message_expiry(
         self, iggy_client: IggyClient, unique_name, invalid_message_expiry
     ):
-        """Test update_topic rejects message_expiry values that are not timedeltas."""
+        """Test update_topic rejects non-IggyExpiry message_expiry values."""
         stream_name = unique_name()
         topic_name = unique_name()
 
@@ -937,7 +945,7 @@ async def test_update_topic_invalid_replication_factor(
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        ("max_topic_size", "expected_exception"),
+        ("max_topic_size_bytes", "expected_exception"),
         [
             (-1, OverflowError),
             (2e64, TypeError),
@@ -947,7 +955,7 @@ async def test_update_topic_invalid_max_topic_size(
         self,
         iggy_client: IggyClient,
         unique_name,
-        max_topic_size,
+        max_topic_size_bytes,
         expected_exception,
     ):
         """Test update_topic rejects invalid maximum topic size values."""
@@ -964,7 +972,7 @@ async def test_update_topic_invalid_max_topic_size(
                 stream_id=stream_name,
                 topic_id=topic_name,
                 name=topic_name,
-                max_topic_size=max_topic_size,
+                max_topic_size=MaxTopicSize.Custom(max_topic_size_bytes),
             )
 
     @pytest.mark.asyncio
@@ -984,7 +992,7 @@ async def test_update_topic_with_message_expiry(
             stream_id=stream_name,
             topic_id=topic_name,
             name=topic_name,
-            message_expiry=timedelta(minutes=10),
+            message_expiry=IggyExpiry.ExpireDuration(timedelta(minutes=10)),
         )
 
         topic = await iggy_client.get_topic(stream_name, topic_name)
@@ -997,19 +1005,16 @@ async def test_update_topic_with_message_expiry(
     @pytest.mark.parametrize(
         ("max_topic_size", "expected_kind"),
         [
-            # Unlike create_topic, update_topic does not resolve the
-            # server-default sentinel (0) to a concrete value, so it round
-            # trips as-is instead of becoming "unlimited".
-            (0, "server_default"),
-            (2_000_000_000, "custom"),
-            (2**64 - 1, "unlimited"),
+            (MaxTopicSize.ServerDefault(), "server_default"),
+            (MaxTopicSize.Custom(2_000_000_000), "custom"),
+            (MaxTopicSize.Unlimited(), "unlimited"),
         ],
     )
     async def test_update_topic_with_valid_max_topic_size(
         self,
         iggy_client: IggyClient,
         unique_name,
-        max_topic_size: int,
+        max_topic_size: MaxTopicSize,
         expected_kind: str,
     ):
         """Test update_topic accepts supported maximum topic size values."""
@@ -1037,7 +1042,8 @@ async def test_update_topic_with_valid_max_topic_size(
             assert isinstance(topic.max_topic_size, MaxTopicSize.Unlimited)
         else:
             assert isinstance(topic.max_topic_size, MaxTopicSize.Custom)
-            assert topic.max_topic_size.bytes == max_topic_size
+            assert isinstance(max_topic_size, MaxTopicSize.Custom)
+            assert topic.max_topic_size.bytes == max_topic_size.bytes
 
     @pytest.mark.asyncio
     async def test_update_topic_applies_repeated_updates(

From 87aabd70dabc6e3efab89f2b3148d8aceacfed5e Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Fri, 10 Jul 2026 19:29:38 -0400
Subject: [PATCH 04/12] docs(python): spell out IggyExpiry/MaxTopicSize variant
 semantics

---
 foreign/python/apache_iggy.pyi | 29 +++++++++++++++++++++++------
 foreign/python/src/client.rs   |  6 ++++--
 foreign/python/src/topic.rs    | 29 +++++++++++++++++++++++------
 3 files changed, 50 insertions(+), 14 deletions(-)

diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index fbf566cef6..c4876bd03b 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -648,7 +648,8 @@ class IggyExpiry:
     @typing.final
     class ServerDefault(IggyExpiry):
         r"""
-        Use the server's default message expiry.
+        Use the message expiry configured on the server for this topic,
+        rather than an explicit value set by the client.
         """
 
         __match_args__ = ()
@@ -659,7 +660,14 @@ class IggyExpiry:
     @typing.final
     class ExpireDuration(IggyExpiry):
         r"""
-        Messages expire after this duration.
+        Expire messages this long after they are appended to the topic.
+
+        `duration` must be greater than zero: a zero-length `timedelta` is
+        indistinguishable on the wire from `ServerDefault` and is treated as
+        such by the server. The upper bound is whatever a `datetime.timedelta`
+        can represent that also fits in a `u64` microsecond count (about
+        584,942 years); in practice the server-configured maximum is reached
+        long before that.
         """
 
         __match_args__ = ("duration",)
@@ -670,7 +678,7 @@ class IggyExpiry:
     @typing.final
     class NeverExpire(IggyExpiry):
         r"""
-        Messages never expire.
+        Retain messages indefinitely; they never expire.
         """
 
         __match_args__ = ()
@@ -687,7 +695,8 @@ class MaxTopicSize:
     @typing.final
     class ServerDefault(MaxTopicSize):
         r"""
-        Use the server's default max size.
+        Use the maximum topic size configured on the server, rather than an
+        explicit value set by the client.
         """
 
         __match_args__ = ()
@@ -698,7 +707,15 @@ class MaxTopicSize:
     @typing.final
     class Custom(MaxTopicSize):
         r"""
-        The topic is limited to this many bytes.
+        Cap the topic at this many bytes; as the topic approaches this size,
+        the server deletes the oldest sealed segments to make room for new
+        messages.
+
+        `bytes` must be greater than zero and less than `u64::MAX`: those two
+        values are reserved on the wire for `ServerDefault` and `Unlimited`
+        respectively, so a `Custom` size at either boundary would be
+        indistinguishable from one of the other variants and is treated as
+        such by the server.
         """
 
         __match_args__ = ("bytes",)
@@ -709,7 +726,7 @@ class MaxTopicSize:
     @typing.final
     class Unlimited(MaxTopicSize):
         r"""
-        The topic has no maximum size.
+        Do not cap the topic size; it may grow without bound.
         """
 
         __match_args__ = ()
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index bfc41f40c8..1e2acfceed 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -203,7 +203,8 @@ impl IggyClient {
 
         let expiry = message_expiry.map_or(RustIggyExpiry::ServerDefault, RustIggyExpiry::from);
 
-        let max_size = max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
+        let max_size =
+            max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
 
         let stream = Identifier::try_from(stream)?;
         let inner = self.inner.clone();
@@ -326,7 +327,8 @@ impl IggyClient {
 
         let expiry = message_expiry.map_or(RustIggyExpiry::ServerDefault, RustIggyExpiry::from);
 
-        let max_size = max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
+        let max_size =
+            max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
 
         let stream_id = Identifier::try_from(stream_id)?;
         let topic_id = Identifier::try_from(topic_id)?;
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index 2c042087a1..147b7acb49 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -31,11 +31,19 @@ use crate::consumer::py_delta_to_iggy_duration;
 #[gen_stub_pyclass_complex_enum]
 #[pyclass]
 pub enum IggyExpiry {
-    /// Use the server's default message expiry.
+    /// Use the message expiry configured on the server for this topic,
+    /// rather than an explicit value set by the client.
     ServerDefault(),
-    /// Messages expire after this duration.
+    /// Expire messages this long after they are appended to the topic.
+    ///
+    /// `duration` must be greater than zero: a zero-length `timedelta` is
+    /// indistinguishable on the wire from `ServerDefault` and is treated as
+    /// such by the server. The upper bound is whatever a `datetime.timedelta`
+    /// can represent that also fits in a `u64` microsecond count (about
+    /// 584,942 years); in practice the server-configured maximum is reached
+    /// long before that.
     ExpireDuration { duration: Py<PyDelta> },
-    /// Messages never expire.
+    /// Retain messages indefinitely; they never expire.
     NeverExpire(),
 }
 
@@ -83,11 +91,20 @@ fn iggy_duration_to_py_delta(duration: Duration) -> Py<PyDelta> {
 #[gen_stub_pyclass_complex_enum]
 #[pyclass]
 pub enum MaxTopicSize {
-    /// Use the server's default max size.
+    /// Use the maximum topic size configured on the server, rather than an
+    /// explicit value set by the client.
     ServerDefault(),
-    /// The topic is limited to this many bytes.
+    /// Cap the topic at this many bytes; as the topic approaches this size,
+    /// the server deletes the oldest sealed segments to make room for new
+    /// messages.
+    ///
+    /// `bytes` must be greater than zero and less than `u64::MAX`: those two
+    /// values are reserved on the wire for `ServerDefault` and `Unlimited`
+    /// respectively, so a `Custom` size at either boundary would be
+    /// indistinguishable from one of the other variants and is treated as
+    /// such by the server.
     Custom { bytes: u64 },
-    /// The topic has no maximum size.
+    /// Do not cap the topic size; it may grow without bound.
     Unlimited(),
 }
 

From b55642f33bddab6c17542cf306693fcdc3a9c77f Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Sun, 12 Jul 2026 12:58:28 -0400
Subject: [PATCH 05/12] fix(python): reject negative timedelta durations

---
 foreign/python/src/client.rs       | 22 ++++++++++++--------
 foreign/python/src/consumer.rs     | 32 ++++++++++++++++++++----------
 foreign/python/src/topic.rs        | 12 ++++++-----
 foreign/python/tests/test_topic.py | 18 +++++++++++++++++
 4 files changed, 60 insertions(+), 24 deletions(-)

diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 1e2acfceed..b3b517ad6b 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -16,8 +16,8 @@
 // under the License.
 
 use iggy::prelude::{
-    Consumer as RustConsumer, IggyClient as RustIggyClient, IggyExpiry as RustIggyExpiry,
-    IggyMessage as RustMessage, MaxTopicSize as RustMaxTopicSize,
+    AutoCommit as RustAutoCommit, Consumer as RustConsumer, IggyClient as RustIggyClient,
+    IggyExpiry as RustIggyExpiry, IggyMessage as RustMessage, MaxTopicSize as RustMaxTopicSize,
     PollingStrategy as RustPollingStrategy, *,
 };
 use pyo3::PyRef;
@@ -201,7 +201,10 @@ impl IggyClient {
             None => CompressionAlgorithm::default(),
         };
 
-        let expiry = message_expiry.map_or(RustIggyExpiry::ServerDefault, RustIggyExpiry::from);
+        let expiry = message_expiry
+            .map(RustIggyExpiry::try_from)
+            .transpose()?
+            .unwrap_or(RustIggyExpiry::ServerDefault);
 
         let max_size =
             max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
@@ -325,7 +328,10 @@ impl IggyClient {
             None => CompressionAlgorithm::default(),
         };
 
-        let expiry = message_expiry.map_or(RustIggyExpiry::ServerDefault, RustIggyExpiry::from);
+        let expiry = message_expiry
+            .map(RustIggyExpiry::try_from)
+            .transpose()?
+            .unwrap_or(RustIggyExpiry::ServerDefault);
 
         let max_size =
             max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
@@ -670,16 +676,16 @@ impl IggyClient {
             builder = builder.batch_length(batch_length)
         };
         if let Some(auto_commit) = auto_commit {
-            builder = builder.auto_commit(auto_commit.into())
+            builder = builder.auto_commit(RustAutoCommit::try_from(auto_commit)?)
         };
         if let Some(poll_interval) = poll_interval {
-            builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval))
+            builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval)?)
         } else {
             builder = builder.without_poll_interval()
         };
         if let Some(polling_retry_interval) = polling_retry_interval {
             builder =
-                builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval))
+                builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval)?)
         }
         if init_retries.is_some() && init_retry_interval.is_none() {
             return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
@@ -695,7 +701,7 @@ impl IggyClient {
         {
             builder = builder.init_retries(
                 init_retries,
-                py_delta_to_iggy_duration(&init_retry_interval),
+                py_delta_to_iggy_duration(&init_retry_interval)?,
             );
         }
         if allow_replay {
diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs
index 0b6066b686..cbf2e99df6 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -27,7 +27,7 @@ use iggy::prelude::{
     ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyDuration,
     IggyError, ReceivedMessage,
 };
-use pyo3::exceptions::PyStopAsyncIteration;
+use pyo3::exceptions::{PyStopAsyncIteration, PyValueError};
 use pyo3::types::{PyDelta, PyDeltaAccess};
 
 use pyo3::prelude::*;
@@ -429,25 +429,27 @@ pub enum AutoCommit {
     After(AutoCommitAfter),
 }
 
-impl From<&AutoCommit> for RustAutoCommit {
-    fn from(val: &AutoCommit) -> RustAutoCommit {
-        match val {
+impl TryFrom<&AutoCommit> for RustAutoCommit {
+    type Error = PyErr;
+
+    fn try_from(val: &AutoCommit) -> PyResult<RustAutoCommit> {
+        Ok(match val {
             AutoCommit::Disabled() => RustAutoCommit::Disabled,
             AutoCommit::Interval(delta) => {
-                let duration = py_delta_to_iggy_duration(delta);
+                let duration = py_delta_to_iggy_duration(delta)?;
                 RustAutoCommit::Interval(duration)
             }
             AutoCommit::IntervalOrWhen(delta, when) => {
-                let duration = py_delta_to_iggy_duration(delta);
+                let duration = py_delta_to_iggy_duration(delta)?;
                 RustAutoCommit::IntervalOrWhen(duration, when.into())
             }
             AutoCommit::IntervalOrAfter(delta, after) => {
-                let duration = py_delta_to_iggy_duration(delta);
+                let duration = py_delta_to_iggy_duration(delta)?;
                 RustAutoCommit::IntervalOrAfter(duration, after.into())
             }
             AutoCommit::When(when) => RustAutoCommit::When(when.into()),
             AutoCommit::After(after) => RustAutoCommit::After(after.into()),
-        }
+        })
     }
 }
 
@@ -517,11 +519,19 @@ impl PyStubType for AutoCommitAfter {
     }
 }
 
-pub fn py_delta_to_iggy_duration(delta1: &Py<PyDelta>) -> IggyDuration {
+pub fn py_delta_to_iggy_duration(delta1: &Py<PyDelta>) -> PyResult<IggyDuration> {
     Python::attach(|py| {
         let delta = delta1.bind(py);
-        let seconds = (delta.get_days() * 60 * 60 * 24 + delta.get_seconds()) as u64;
+        let total_seconds = delta.get_days() * 60 * 60 * 24 + delta.get_seconds();
+        if total_seconds < 0 {
+            return Err(PyValueError::new_err(
+                "duration must not be negative".to_string(),
+            ));
+        }
         let nanos = (delta.get_microseconds() * 1_000) as u32;
-        IggyDuration::new(Duration::new(seconds, nanos))
+        Ok(IggyDuration::new(Duration::new(
+            total_seconds as u64,
+            nanos,
+        )))
     })
 }
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index 147b7acb49..3e14b35190 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -59,15 +59,17 @@ impl From<RustIggyExpiry> for IggyExpiry {
     }
 }
 
-impl From<&IggyExpiry> for RustIggyExpiry {
-    fn from(expiry: &IggyExpiry) -> Self {
-        match expiry {
+impl TryFrom<&IggyExpiry> for RustIggyExpiry {
+    type Error = PyErr;
+
+    fn try_from(expiry: &IggyExpiry) -> PyResult<Self> {
+        Ok(match expiry {
             IggyExpiry::ServerDefault() => RustIggyExpiry::ServerDefault,
             IggyExpiry::ExpireDuration { duration } => {
-                RustIggyExpiry::ExpireDuration(py_delta_to_iggy_duration(duration))
+                RustIggyExpiry::ExpireDuration(py_delta_to_iggy_duration(duration)?)
             }
             IggyExpiry::NeverExpire() => RustIggyExpiry::NeverExpire,
-        }
+        })
     }
 }
 
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 47658e8785..204e41ace0 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -269,6 +269,24 @@ async def test_create_topic_with_message_expiry(
             assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
             assert topic.message_expiry.duration == message_expiry.duration
 
+    @pytest.mark.asyncio
+    async def test_create_topic_rejects_negative_message_expiry(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test create_topic rejects a negative ExpireDuration timedelta."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+
+        await iggy_client.create_stream(stream_name)
+
+        with pytest.raises(ValueError):
+            await iggy_client.create_topic(
+                stream=stream_name,
+                name=topic_name,
+                partitions_count=1,
+                message_expiry=IggyExpiry.ExpireDuration(timedelta(seconds=-1)),
+            )
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         "invalid_message_expiry", [1, "1s", object(), timedelta(seconds=1)]

From 78b65000d8a2889d44d239811a8afd73328e3cd5 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Sun, 12 Jul 2026 19:28:46 -0400
Subject: [PATCH 06/12] docs(python): note ValueError on negative message
 expiry

---
 foreign/python/apache_iggy.pyi | 9 +++++----
 foreign/python/src/topic.rs    | 9 +++++----
 2 files changed, 10 insertions(+), 8 deletions(-)

diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index c4876bd03b..499673c1f6 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -664,10 +664,11 @@ class IggyExpiry:
 
         `duration` must be greater than zero: a zero-length `timedelta` is
         indistinguishable on the wire from `ServerDefault` and is treated as
-        such by the server. The upper bound is whatever a `datetime.timedelta`
-        can represent that also fits in a `u64` microsecond count (about
-        584,942 years); in practice the server-configured maximum is reached
-        long before that.
+        such by the server. A negative `timedelta` raises `ValueError` when
+        this value is passed to `create_topic`/`update_topic`. The upper
+        bound is whatever a `datetime.timedelta` can represent that also fits
+        in a `u64` microsecond count (about 584,942 years); in practice the
+        server-configured maximum is reached long before that.
         """
 
         __match_args__ = ("duration",)
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index 3e14b35190..a2c9820db2 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -38,10 +38,11 @@ pub enum IggyExpiry {
     ///
     /// `duration` must be greater than zero: a zero-length `timedelta` is
     /// indistinguishable on the wire from `ServerDefault` and is treated as
-    /// such by the server. The upper bound is whatever a `datetime.timedelta`
-    /// can represent that also fits in a `u64` microsecond count (about
-    /// 584,942 years); in practice the server-configured maximum is reached
-    /// long before that.
+    /// such by the server. A negative `timedelta` raises `ValueError` when
+    /// this value is passed to `create_topic`/`update_topic`. The upper
+    /// bound is whatever a `datetime.timedelta` can represent that also fits
+    /// in a `u64` microsecond count (about 584,942 years); in practice the
+    /// server-configured maximum is reached long before that.
     ExpireDuration { duration: Py<PyDelta> },
     /// Retain messages indefinitely; they never expire.
     NeverExpire(),

From a511223fa32a756649459d510a0ca917e8e24b26 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Sun, 12 Jul 2026 19:30:58 -0400
Subject: [PATCH 07/12] fix(python): reject max topic size at u64 boundary

---
 foreign/python/apache_iggy.pyi     | 17 ++++++++++-------
 foreign/python/src/client.rs       | 19 +++++++++++++------
 foreign/python/src/topic.rs        | 30 ++++++++++++++++++++----------
 foreign/python/tests/test_topic.py |  2 ++
 4 files changed, 45 insertions(+), 23 deletions(-)

diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 499673c1f6..e00dc9b022 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -367,7 +367,9 @@ class IggyClient:
     ) -> collections.abc.Awaitable[None]:
         r"""
         Creates a new topic with the given parameters.
-        Returns Ok(()) on successful topic creation or a PyRuntimeError on failure.
+        Returns Ok(()) on successful topic creation, raises ValueError for an
+        invalid `message_expiry` or `max_topic_size`, or a PyRuntimeError on
+        other failures.
         """
     def get_topic(
         self,
@@ -422,7 +424,8 @@ class IggyClient:
             An awaitable that resolves to `None` when the topic is updated.
 
         Raises:
-            PyRuntimeError: If an argument is invalid or the request fails.
+            ValueError: If `message_expiry` or `max_topic_size` is out of range.
+            PyRuntimeError: If another argument is invalid or the request fails.
         """
     def delete_topic(
         self,
@@ -712,11 +715,11 @@ class MaxTopicSize:
         the server deletes the oldest sealed segments to make room for new
         messages.
 
-        `bytes` must be greater than zero and less than `u64::MAX`: those two
-        values are reserved on the wire for `ServerDefault` and `Unlimited`
-        respectively, so a `Custom` size at either boundary would be
-        indistinguishable from one of the other variants and is treated as
-        such by the server.
+        `bytes` must be greater than zero and less than the maximum value of
+        an unsigned 64-bit integer: those two values are reserved on the wire
+        for `ServerDefault` and `Unlimited` respectively, so a `Custom` size
+        at either boundary raises `ValueError` when passed to
+        `create_topic`/`update_topic`.
         """
 
         __match_args__ = ("bytes",)
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index b3b517ad6b..127e2d1c43 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -170,7 +170,9 @@ impl IggyClient {
     }
 
     /// Creates a new topic with the given parameters.
-    /// Returns Ok(()) on successful topic creation or a PyRuntimeError on failure.
+    /// Returns Ok(()) on successful topic creation, raises ValueError for an
+    /// invalid `message_expiry` or `max_topic_size`, or a PyRuntimeError on
+    /// other failures.
     #[pyo3(
         signature = (stream, name, partitions_count, compression_algorithm = None, replication_factor = None, message_expiry = None, max_topic_size = None)
     )]
@@ -206,8 +208,10 @@ impl IggyClient {
             .transpose()?
             .unwrap_or(RustIggyExpiry::ServerDefault);
 
-        let max_size =
-            max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
+        let max_size = max_topic_size
+            .map(RustMaxTopicSize::try_from)
+            .transpose()?
+            .unwrap_or(RustMaxTopicSize::ServerDefault);
 
         let stream = Identifier::try_from(stream)?;
         let inner = self.inner.clone();
@@ -297,7 +301,8 @@ impl IggyClient {
     ///     An awaitable that resolves to `None` when the topic is updated.
     ///
     /// Raises:
-    ///     PyRuntimeError: If an argument is invalid or the request fails.
+    ///     ValueError: If `message_expiry` or `max_topic_size` is out of range.
+    ///     PyRuntimeError: If another argument is invalid or the request fails.
     #[pyo3(
         signature = (stream_id, topic_id, name, compression_algorithm = None, replication_factor = None, message_expiry = None, max_topic_size = None)
     )]
@@ -333,8 +338,10 @@ impl IggyClient {
             .transpose()?
             .unwrap_or(RustIggyExpiry::ServerDefault);
 
-        let max_size =
-            max_topic_size.map_or(RustMaxTopicSize::ServerDefault, RustMaxTopicSize::from);
+        let max_size = max_topic_size
+            .map(RustMaxTopicSize::try_from)
+            .transpose()?
+            .unwrap_or(RustMaxTopicSize::ServerDefault);
 
         let stream_id = Identifier::try_from(stream_id)?;
         let topic_id = Identifier::try_from(topic_id)?;
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index a2c9820db2..d848a14ed1 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -21,6 +21,7 @@ use iggy::prelude::{
     IggyByteSize, IggyExpiry as RustIggyExpiry, MaxTopicSize as RustMaxTopicSize,
     Partition as RustPartition, Topic as RustTopic, TopicDetails as RustTopicDetails,
 };
+use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::PyDelta;
 use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum, gen_stub_pymethods};
@@ -101,11 +102,11 @@ pub enum MaxTopicSize {
     /// the server deletes the oldest sealed segments to make room for new
     /// messages.
     ///
-    /// `bytes` must be greater than zero and less than `u64::MAX`: those two
-    /// values are reserved on the wire for `ServerDefault` and `Unlimited`
-    /// respectively, so a `Custom` size at either boundary would be
-    /// indistinguishable from one of the other variants and is treated as
-    /// such by the server.
+    /// `bytes` must be greater than zero and less than the maximum value of
+    /// an unsigned 64-bit integer: those two values are reserved on the wire
+    /// for `ServerDefault` and `Unlimited` respectively, so a `Custom` size
+    /// at either boundary raises `ValueError` when passed to
+    /// `create_topic`/`update_topic`.
     Custom { bytes: u64 },
     /// Do not cap the topic size; it may grow without bound.
     Unlimited(),
@@ -123,13 +124,22 @@ impl From<RustMaxTopicSize> for MaxTopicSize {
     }
 }
 
-impl From<&MaxTopicSize> for RustMaxTopicSize {
-    fn from(max_size: &MaxTopicSize) -> Self {
-        match max_size {
+impl TryFrom<&MaxTopicSize> for RustMaxTopicSize {
+    type Error = PyErr;
+
+    fn try_from(max_size: &MaxTopicSize) -> PyResult<Self> {
+        Ok(match max_size {
             MaxTopicSize::ServerDefault() => RustMaxTopicSize::ServerDefault,
-            MaxTopicSize::Custom { bytes } => RustMaxTopicSize::Custom(IggyByteSize::from(*bytes)),
+            MaxTopicSize::Custom { bytes } => {
+                if *bytes == 0 || *bytes == u64::MAX {
+                    return Err(PyValueError::new_err(
+                        "bytes must be greater than zero and less than u64::MAX".to_string(),
+                    ));
+                }
+                RustMaxTopicSize::Custom(IggyByteSize::from(*bytes))
+            }
             MaxTopicSize::Unlimited() => RustMaxTopicSize::Unlimited,
-        }
+        })
     }
 }
 
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 204e41ace0..eccbe707c1 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -353,6 +353,8 @@ async def test_create_topic_with_valid_max_topic_size(
             (4563, RuntimeError),
             (-1, OverflowError),
             (2e64, TypeError),
+            (0, ValueError),
+            (2**64 - 1, ValueError),  # u64::MAX is reserved for Unlimited
         ],
     )
     async def test_create_topic_invalid_max_topic_size(

From d44155e8f9d55916b2f28139f70b243b7e581998 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@Matthews-MacBook-Pro.local>
Date: Mon, 20 Jul 2026 10:07:38 -0400
Subject: [PATCH 08/12] fix(python): widen timedelta-to-seconds conversion to
 i64

---
 foreign/python/src/consumer.rs     | 3 ++-
 foreign/python/src/topic.rs        | 2 +-
 foreign/python/tests/test_topic.py | 3 +++
 3 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs
index cbf2e99df6..f9b267d2c9 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -522,7 +522,8 @@ impl PyStubType for AutoCommitAfter {
 pub fn py_delta_to_iggy_duration(delta1: &Py<PyDelta>) -> PyResult<IggyDuration> {
     Python::attach(|py| {
         let delta = delta1.bind(py);
-        let total_seconds = delta.get_days() * 60 * 60 * 24 + delta.get_seconds();
+        let total_seconds =
+            i64::from(delta.get_days()) * 86_400 + i64::from(delta.get_seconds());
         if total_seconds < 0 {
             return Err(PyValueError::new_err(
                 "duration must not be negative".to_string(),
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index d848a14ed1..b03a8285e8 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -42,7 +42,7 @@ pub enum IggyExpiry {
     /// such by the server. A negative `timedelta` raises `ValueError` when
     /// this value is passed to `create_topic`/`update_topic`. The upper
     /// bound is whatever a `datetime.timedelta` can represent that also fits
-    /// in a `u64` microsecond count (about 584,942 years); in practice the
+    /// in a `u64` microsecond count (about 584,542 years); in practice the
     /// server-configured maximum is reached long before that.
     ExpireDuration { duration: Py<PyDelta> },
     /// Retain messages indefinitely; they never expire.
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index eccbe707c1..3fcf2928c4 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -238,6 +238,9 @@ async def test_create_topic_invalid_compression_algorithm(
             IggyExpiry.ExpireDuration(timedelta(seconds=1)),
             IggyExpiry.ExpireDuration(timedelta(minutes=10)),
             IggyExpiry.ExpireDuration(timedelta(days=1, seconds=2, microseconds=3)),
+            # days * 86_400 overflows i32 (max ~24,855 days); regression test
+            # for widening the days-to-seconds conversion to i64.
+            IggyExpiry.ExpireDuration(timedelta(days=30_000)),
         ],
     )
     async def test_create_topic_with_message_expiry(

From 529411f49b06dd2a7a3c65b14a927e62f7ab01a0 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@Matthews-MacBook-Pro.local>
Date: Mon, 20 Jul 2026 11:32:11 -0400
Subject: [PATCH 09/12] fix(python): reject IggyExpiry sentinel-boundary
 durations

---
 foreign/python/src/topic.rs        | 32 ++++++++++++++++++++++--------
 foreign/python/tests/test_topic.py | 30 +++++++++++++++++-----------
 2 files changed, 42 insertions(+), 20 deletions(-)

diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index b03a8285e8..c15252fbd7 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -37,13 +37,12 @@ pub enum IggyExpiry {
     ServerDefault(),
     /// Expire messages this long after they are appended to the topic.
     ///
-    /// `duration` must be greater than zero: a zero-length `timedelta` is
-    /// indistinguishable on the wire from `ServerDefault` and is treated as
-    /// such by the server. A negative `timedelta` raises `ValueError` when
-    /// this value is passed to `create_topic`/`update_topic`. The upper
-    /// bound is whatever a `datetime.timedelta` can represent that also fits
-    /// in a `u64` microsecond count (about 584,542 years); in practice the
-    /// server-configured maximum is reached long before that.
+    /// `duration` must be greater than zero and less than the maximum
+    /// microsecond count a `u64` can hold (about 584,542 years): those two
+    /// values are reserved on the wire for `ServerDefault` and `NeverExpire`
+    /// respectively, so a `duration` at either boundary raises `ValueError`
+    /// when passed to `create_topic`/`update_topic`. A negative `timedelta`
+    /// also raises `ValueError`.
     ExpireDuration { duration: Py<PyDelta> },
     /// Retain messages indefinitely; they never expire.
     NeverExpire(),
@@ -68,7 +67,24 @@ impl TryFrom<&IggyExpiry> for RustIggyExpiry {
         Ok(match expiry {
             IggyExpiry::ServerDefault() => RustIggyExpiry::ServerDefault,
             IggyExpiry::ExpireDuration { duration } => {
-                RustIggyExpiry::ExpireDuration(py_delta_to_iggy_duration(duration)?)
+                let iggy_duration = py_delta_to_iggy_duration(duration)?;
+                if iggy_duration.is_zero() {
+                    return Err(PyValueError::new_err(
+                        "duration must be greater than zero and less than the maximum \
+                         representable microsecond count; those values are reserved for \
+                         IggyExpiry.ServerDefault() and IggyExpiry.NeverExpire() respectively"
+                            .to_string(),
+                    ));
+                }
+                if iggy_duration.get_duration().as_micros() >= u64::MAX as u128 {
+                    return Err(PyValueError::new_err(
+                        "duration must be greater than zero and less than the maximum \
+                         representable microsecond count; those values are reserved for \
+                         IggyExpiry.ServerDefault() and IggyExpiry.NeverExpire() respectively"
+                            .to_string(),
+                    ));
+                }
+                RustIggyExpiry::ExpireDuration(iggy_duration)
             }
             IggyExpiry::NeverExpire() => RustIggyExpiry::NeverExpire,
         })
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 3fcf2928c4..9772d095ce 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -233,7 +233,6 @@ async def test_create_topic_invalid_compression_algorithm(
     @pytest.mark.parametrize(
         "message_expiry",
         [
-            IggyExpiry.ExpireDuration(timedelta(0)),  # server default sentinel
             IggyExpiry.ExpireDuration(timedelta(microseconds=1)),
             IggyExpiry.ExpireDuration(timedelta(seconds=1)),
             IggyExpiry.ExpireDuration(timedelta(minutes=10)),
@@ -264,19 +263,26 @@ async def test_create_topic_with_message_expiry(
         topic = await iggy_client.get_topic(stream_name, topic_name)
         assert topic is not None
         assert topic.name == topic_name
-        if message_expiry.duration == timedelta(0):
-            # A zero duration is the server-default sentinel; the server
-            # resolves it to the configured default, which is "never expire".
-            assert isinstance(topic.message_expiry, IggyExpiry.NeverExpire)
-        else:
-            assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
-            assert topic.message_expiry.duration == message_expiry.duration
+        assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
+        assert topic.message_expiry.duration == message_expiry.duration
 
     @pytest.mark.asyncio
-    async def test_create_topic_rejects_negative_message_expiry(
-        self, iggy_client: IggyClient, unique_name
+    @pytest.mark.parametrize(
+        "invalid_duration",
+        [
+            timedelta(seconds=-1),
+            # 0 is the wire sentinel reserved for IggyExpiry.ServerDefault();
+            # matches MaxTopicSize.Custom(0) rejecting its own sentinel.
+            timedelta(0),
+            # u64::MAX microseconds is the wire sentinel reserved for
+            # IggyExpiry.NeverExpire(); matches MaxTopicSize.Custom(u64::MAX).
+            timedelta(microseconds=2**64 - 1),
+        ],
+    )
+    async def test_create_topic_rejects_invalid_message_expiry_duration(
+        self, iggy_client: IggyClient, unique_name, invalid_duration: timedelta
     ):
-        """Test create_topic rejects a negative ExpireDuration timedelta."""
+        """Test create_topic rejects an ExpireDuration at a reserved boundary."""
         stream_name = unique_name()
         topic_name = unique_name()
 
@@ -287,7 +293,7 @@ async def test_create_topic_rejects_negative_message_expiry(
                 stream=stream_name,
                 name=topic_name,
                 partitions_count=1,
-                message_expiry=IggyExpiry.ExpireDuration(timedelta(seconds=-1)),
+                message_expiry=IggyExpiry.ExpireDuration(invalid_duration),
             )
 
     @pytest.mark.asyncio

From 9f8445b8114541cc868c30042bace7dfdc8531a9 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@Matthews-MacBook-Pro.local>
Date: Mon, 20 Jul 2026 13:57:01 -0400
Subject: [PATCH 10/12] refactor(python): dedup topic param resolution, mirror
 update_topic tests

---
 foreign/python/apache_iggy.pyi     | 35 +++++++++----
 foreign/python/src/client.rs       | 79 +++++++++++++++++-------------
 foreign/python/src/topic.rs        |  3 ++
 foreign/python/tests/test_topic.py | 73 +++++++++++++++++++++++++++
 4 files changed, 147 insertions(+), 43 deletions(-)

diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index e00dc9b022..ddf43bcc20 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -367,9 +367,22 @@ class IggyClient:
     ) -> collections.abc.Awaitable[None]:
         r"""
         Creates a new topic with the given parameters.
-        Returns Ok(()) on successful topic creation, raises ValueError for an
-        invalid `message_expiry` or `max_topic_size`, or a PyRuntimeError on
-        other failures.
+
+        Args:
+            stream: Stream identifier as `str | int`.
+            name: Topic name as `str`.
+            partitions_count: Number of partitions as `int`.
+            compression_algorithm: Compression algorithm as `str | None`.
+            replication_factor: Replication factor as `int | None`.
+            message_expiry: Message expiry as `IggyExpiry | None`.
+            max_topic_size: Maximum topic size as `MaxTopicSize | None`.
+
+        Returns:
+            An awaitable that resolves to `None` when the topic is created.
+
+        Raises:
+            ValueError: If `message_expiry` or `max_topic_size` is out of range.
+            PyRuntimeError: If another argument is invalid or the request fails.
         """
     def get_topic(
         self,
@@ -665,13 +678,12 @@ class IggyExpiry:
         r"""
         Expire messages this long after they are appended to the topic.
 
-        `duration` must be greater than zero: a zero-length `timedelta` is
-        indistinguishable on the wire from `ServerDefault` and is treated as
-        such by the server. A negative `timedelta` raises `ValueError` when
-        this value is passed to `create_topic`/`update_topic`. The upper
-        bound is whatever a `datetime.timedelta` can represent that also fits
-        in a `u64` microsecond count (about 584,942 years); in practice the
-        server-configured maximum is reached long before that.
+        `duration` must be greater than zero and less than the maximum
+        microsecond count a `u64` can hold (about 584,542 years): those two
+        values are reserved on the wire for `ServerDefault` and `NeverExpire`
+        respectively, so a `duration` at either boundary raises `ValueError`
+        when passed to `create_topic`/`update_topic`. A negative `timedelta`
+        also raises `ValueError`.
         """
 
         __match_args__ = ("duration",)
@@ -980,4 +992,7 @@ class TopicDetails:
     def partitions(self) -> builtins.list[Partition]:
         r"""
         The collection of partitions in the topic.
+
+        Rebuilds the list from scratch on every access; cache the result
+        rather than reading this repeatedly in a loop.
         """
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 127e2d1c43..c70f3dc491 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -49,6 +49,32 @@ pub struct IggyClient {
     inner: Arc<RustIggyClient>,
 }
 
+/// Resolves the shared `create_topic`/`update_topic` parameters, applying
+/// server defaults where the caller left them unset.
+fn resolve_topic_params(
+    compression_algorithm: Option<String>,
+    message_expiry: Option<&IggyExpiry>,
+    max_topic_size: Option<&MaxTopicSize>,
+) -> PyResult<(CompressionAlgorithm, RustIggyExpiry, RustMaxTopicSize)> {
+    let compression_algorithm = match compression_algorithm {
+        Some(algo) => CompressionAlgorithm::from_str(&algo)
+            .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
+        None => CompressionAlgorithm::default(),
+    };
+
+    let expiry = message_expiry
+        .map(RustIggyExpiry::try_from)
+        .transpose()?
+        .unwrap_or(RustIggyExpiry::ServerDefault);
+
+    let max_size = max_topic_size
+        .map(RustMaxTopicSize::try_from)
+        .transpose()?
+        .unwrap_or(RustMaxTopicSize::ServerDefault);
+
+    Ok((compression_algorithm, expiry, max_size))
+}
+
 #[gen_stub_pymethods]
 #[pymethods]
 impl IggyClient {
@@ -170,9 +196,22 @@ impl IggyClient {
     }
 
     /// Creates a new topic with the given parameters.
-    /// Returns Ok(()) on successful topic creation, raises ValueError for an
-    /// invalid `message_expiry` or `max_topic_size`, or a PyRuntimeError on
-    /// other failures.
+    ///
+    /// Args:
+    ///     stream: Stream identifier as `str | int`.
+    ///     name: Topic name as `str`.
+    ///     partitions_count: Number of partitions as `int`.
+    ///     compression_algorithm: Compression algorithm as `str | None`.
+    ///     replication_factor: Replication factor as `int | None`.
+    ///     message_expiry: Message expiry as `IggyExpiry | None`.
+    ///     max_topic_size: Maximum topic size as `MaxTopicSize | None`.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `None` when the topic is created.
+    ///
+    /// Raises:
+    ///     ValueError: If `message_expiry` or `max_topic_size` is out of range.
+    ///     PyRuntimeError: If another argument is invalid or the request fails.
     #[pyo3(
         signature = (stream, name, partitions_count, compression_algorithm = None, replication_factor = None, message_expiry = None, max_topic_size = None)
     )]
@@ -197,21 +236,8 @@ impl IggyClient {
             &MaxTopicSize,
         >,
     ) -> PyResult<Bound<'a, PyAny>> {
-        let compression_algorithm = match compression_algorithm {
-            Some(algo) => CompressionAlgorithm::from_str(&algo)
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
-            None => CompressionAlgorithm::default(),
-        };
-
-        let expiry = message_expiry
-            .map(RustIggyExpiry::try_from)
-            .transpose()?
-            .unwrap_or(RustIggyExpiry::ServerDefault);
-
-        let max_size = max_topic_size
-            .map(RustMaxTopicSize::try_from)
-            .transpose()?
-            .unwrap_or(RustMaxTopicSize::ServerDefault);
+        let (compression_algorithm, expiry, max_size) =
+            resolve_topic_params(compression_algorithm, message_expiry, max_topic_size)?;
 
         let stream = Identifier::try_from(stream)?;
         let inner = self.inner.clone();
@@ -327,21 +353,8 @@ impl IggyClient {
             &MaxTopicSize,
         >,
     ) -> PyResult<Bound<'a, PyAny>> {
-        let compression_algorithm = match compression_algorithm {
-            Some(algo) => CompressionAlgorithm::from_str(&algo)
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
-            None => CompressionAlgorithm::default(),
-        };
-
-        let expiry = message_expiry
-            .map(RustIggyExpiry::try_from)
-            .transpose()?
-            .unwrap_or(RustIggyExpiry::ServerDefault);
-
-        let max_size = max_topic_size
-            .map(RustMaxTopicSize::try_from)
-            .transpose()?
-            .unwrap_or(RustMaxTopicSize::ServerDefault);
+        let (compression_algorithm, expiry, max_size) =
+            resolve_topic_params(compression_algorithm, message_expiry, max_topic_size)?;
 
         let stream_id = Identifier::try_from(stream_id)?;
         let topic_id = Identifier::try_from(topic_id)?;
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index c15252fbd7..e16ee71796 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -313,6 +313,9 @@ impl TopicDetails {
     }
 
     /// The collection of partitions in the topic.
+    ///
+    /// Rebuilds the list from scratch on every access; cache the result
+    /// rather than reading this repeatedly in a loop.
     #[getter]
     pub fn partitions(&self) -> Vec<Partition> {
         self.inner
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 9772d095ce..678f95c1c1 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -767,6 +767,44 @@ async def test_get_topics_requires_connection_and_auth(self, unique_name):
 class TestUpdateTopic:
     """Test updating topics via update_topic."""
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("prefix", "min_bytes", "max_bytes"),
+        [
+            ("", 0, 0),
+            ("a" * 248, 256, 256),
+            ("é" * 124, 256, 256),
+            (("é" * 123) + "ab", 256, 256),
+            (("한" * 82) + "ab", 256, 256),
+            (("漢" * 82) + "ab", 256, 256),
+            (("あ" * 82) + "ab", 256, 256),
+            ("😀" * 62, 256, 256),
+            (("😀" * 61) + "abcd", 256, 256),
+        ],
+    )
+    async def test_update_topic_invalid_names(
+        self,
+        iggy_client: IggyClient,
+        unique_name,
+        prefix: str,
+        min_bytes: int,
+        max_bytes: int,
+    ):
+        """Test update_topic enforces byte-length validation."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        invalid_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes)
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name, name=topic_name, partitions_count=1
+        )
+
+        with pytest.raises(RuntimeError):
+            await iggy_client.update_topic(
+                stream_id=stream_name, topic_id=topic_name, name=invalid_name
+            )
+
     @pytest.mark.asyncio
     async def test_update_topic_renames_topic(
         self, iggy_client: IggyClient, unique_name
@@ -978,6 +1016,8 @@ async def test_update_topic_invalid_replication_factor(
         [
             (-1, OverflowError),
             (2e64, TypeError),
+            (0, ValueError),
+            (2**64 - 1, ValueError),  # u64::MAX is reserved for Unlimited
         ],
     )
     async def test_update_topic_invalid_max_topic_size(
@@ -1030,6 +1070,39 @@ async def test_update_topic_with_message_expiry(
         assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
         assert topic.message_expiry.duration == timedelta(minutes=10)
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "invalid_duration",
+        [
+            timedelta(seconds=-1),
+            # 0 is the wire sentinel reserved for IggyExpiry.ServerDefault();
+            # matches MaxTopicSize.Custom(0) rejecting its own sentinel.
+            timedelta(0),
+            # u64::MAX microseconds is the wire sentinel reserved for
+            # IggyExpiry.NeverExpire(); matches MaxTopicSize.Custom(u64::MAX).
+            timedelta(microseconds=2**64 - 1),
+        ],
+    )
+    async def test_update_topic_rejects_invalid_message_expiry_duration(
+        self, iggy_client: IggyClient, unique_name, invalid_duration: timedelta
+    ):
+        """Test update_topic rejects an ExpireDuration at a reserved boundary."""
+        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=1
+        )
+
+        with pytest.raises(ValueError):
+            await iggy_client.update_topic(
+                stream_id=stream_name,
+                topic_id=topic_name,
+                name=topic_name,
+                message_expiry=IggyExpiry.ExpireDuration(invalid_duration),
+            )
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         ("max_topic_size", "expected_kind"),

From 9610f27015cb833d4353ab03fdc5fcf7362e0e84 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@Matthews-MacBook-Pro.local>
Date: Mon, 20 Jul 2026 14:07:38 -0400
Subject: [PATCH 11/12] style(python): rustfmt py_delta_to_iggy_duration

---
 foreign/python/src/consumer.rs | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs
index f9b267d2c9..4d64fc626c 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -522,8 +522,7 @@ impl PyStubType for AutoCommitAfter {
 pub fn py_delta_to_iggy_duration(delta1: &Py<PyDelta>) -> PyResult<IggyDuration> {
     Python::attach(|py| {
         let delta = delta1.bind(py);
-        let total_seconds =
-            i64::from(delta.get_days()) * 86_400 + i64::from(delta.get_seconds());
+        let total_seconds = i64::from(delta.get_days()) * 86_400 + i64::from(delta.get_seconds());
         if total_seconds < 0 {
             return Err(PyValueError::new_err(
                 "duration must not be negative".to_string(),

From 320135af6f0328cbc787f858c90b6919ae11f8f3 Mon Sep 17 00:00:00 2001
From: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Date: Wed, 22 Jul 2026 11:19:02 -0400
Subject: [PATCH 12/12] fix(python): raise ValueError on unrepresentable
 message expiry duration

---
 foreign/python/src/topic.rs | 30 ++++++++++++++++++------------
 1 file changed, 18 insertions(+), 12 deletions(-)

diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index e16ee71796..178f90dd5e 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -48,15 +48,17 @@ pub enum IggyExpiry {
     NeverExpire(),
 }
 
-impl From<RustIggyExpiry> for IggyExpiry {
-    fn from(expiry: RustIggyExpiry) -> Self {
-        match expiry {
+impl TryFrom<RustIggyExpiry> for IggyExpiry {
+    type Error = PyErr;
+
+    fn try_from(expiry: RustIggyExpiry) -> PyResult<Self> {
+        Ok(match expiry {
             RustIggyExpiry::ServerDefault => IggyExpiry::ServerDefault(),
             RustIggyExpiry::ExpireDuration(duration) => IggyExpiry::ExpireDuration {
-                duration: iggy_duration_to_py_delta(duration.get_duration()),
+                duration: iggy_duration_to_py_delta(duration.get_duration())?,
             },
             RustIggyExpiry::NeverExpire => IggyExpiry::NeverExpire(),
-        }
+        })
     }
 }
 
@@ -91,7 +93,7 @@ impl TryFrom<&IggyExpiry> for RustIggyExpiry {
     }
 }
 
-fn iggy_duration_to_py_delta(duration: Duration) -> Py<PyDelta> {
+fn iggy_duration_to_py_delta(duration: Duration) -> PyResult<Py<PyDelta>> {
     let days = duration.as_secs() / 86_400;
     let secs_of_day = duration.as_secs() % 86_400;
     Python::attach(|py| {
@@ -102,8 +104,12 @@ fn iggy_duration_to_py_delta(duration: Duration) -> Py<PyDelta> {
             duration.subsec_micros() as i32,
             true,
         )
-        .expect("topic message expiry duration fits within timedelta bounds")
-        .unbind()
+        .map(|delta| delta.unbind())
+        .map_err(|err| {
+            PyValueError::new_err(format!(
+                "topic message expiry duration does not fit within timedelta bounds: {err}"
+            ))
+        })
     })
 }
 
@@ -212,8 +218,8 @@ impl Topic {
 
     /// The expiry of the messages in the topic.
     #[getter]
-    pub fn message_expiry(&self) -> IggyExpiry {
-        self.inner.message_expiry.into()
+    pub fn message_expiry(&self) -> PyResult<IggyExpiry> {
+        self.inner.message_expiry.try_into()
     }
 
     /// Compression algorithm for the topic.
@@ -290,8 +296,8 @@ impl TopicDetails {
 
     /// The expiry of the messages in the topic.
     #[getter]
-    pub fn message_expiry(&self) -> IggyExpiry {
-        self.inner.message_expiry.into()
+    pub fn message_expiry(&self) -> PyResult<IggyExpiry> {
+        self.inner.message_expiry.try_into()
     }
 
     /// Compression algorithm for the topic.