This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 56b961285 feat(python): expose remaining Topic/TopicDetails fields and
partitions (#3623)
56b961285 is described below
commit 56b961285bf3bcf08a27347b321113b39cb03860
Author: Matthew Patton <[email protected]>
AuthorDate: Wed Aug 5 05:51:30 2026 -0400
feat(python): expose remaining Topic/TopicDetails fields and partitions
(#3623)
---
bdd/python/uv.lock | 4 +-
core/common/src/types/partition/mod.rs | 2 +-
examples/python/uv.lock | 2 +-
foreign/python/Cargo.toml | 2 +-
foreign/python/apache_iggy.pyi | 221 +++++++++++++++++++++++++-
foreign/python/pyproject.toml | 2 +-
foreign/python/src/client.rs | 109 ++++++++-----
foreign/python/src/consumer.rs | 32 ++--
foreign/python/src/lib.rs | 5 +-
foreign/python/src/topic.rs | 276 ++++++++++++++++++++++++++++++++-
foreign/python/tests/test_topic.py | 235 ++++++++++++++++++++++++----
foreign/python/uv.lock | 4 +-
12 files changed, 794 insertions(+), 100 deletions(-)
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index ed686c9bd..846067ffc 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.8.1.dev3"
+version = "0.8.1.dev4"
source = { directory = "../../foreign/python" }
[package.metadata]
@@ -200,7 +200,7 @@ name = "exceptiongroup"
version = "1.3.1"
source = { registry = "https://pypi.org/simple"; }
dependencies = [
- { name = "typing-extensions" },
+ { name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url =
"https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz";,
hash =
"sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size
= 30371, upload-time = "2025-11-21T23:01:54.787Z" }
wheels = [
diff --git a/core/common/src/types/partition/mod.rs
b/core/common/src/types/partition/mod.rs
index 7b75151a9..fdd7161ab 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/examples/python/uv.lock b/examples/python/uv.lock
index 639a0f563..a33c0238f 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.8.1.dev3"
+version = "0.8.1.dev4"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index d408c0303..cd9c4eda9 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "apache-iggy"
-version = "0.8.1-dev3"
+version = "0.8.1-dev4"
edition = "2024"
authors = ["Iggy Committers <[email protected]>"]
license = "Apache-2.0"
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 0ce240163..e10a5ec89 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -37,6 +37,9 @@ __all__ = [
"GlobalPermissions",
"IggyClient",
"IggyConsumer",
+ "IggyExpiry",
+ "MaxTopicSize",
+ "Partition",
"Permissions",
"PollingStrategy",
"ReceiveMessage",
@@ -956,12 +959,27 @@ 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.
- Returns Ok(()) on successful topic creation or a PyRuntimeError on
failure.
+
+ 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,
@@ -994,8 +1012,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.
@@ -1009,14 +1027,15 @@ 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.
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,
@@ -1325,6 +1344,134 @@ 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 message expiry configured on the server for this topic,
+ rather than an explicit value set by the client.
+ """
+
+ __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"""
+ Expire messages this long after they are appended to the topic.
+
+ `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",)
+ @property
+ def duration(self) -> datetime.timedelta: ...
+ def __new__(cls, duration: datetime.timedelta) ->
IggyExpiry.ExpireDuration: ...
+
+ @typing.final
+ class NeverExpire(IggyExpiry):
+ r"""
+ Retain messages indefinitely; they 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 maximum topic size configured on the server, rather than an
+ explicit value set by the client.
+ """
+
+ __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"""
+ 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 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",)
+ @property
+ def bytes(self) -> builtins.int: ...
+ def __new__(cls, bytes: builtins.int) -> MaxTopicSize.Custom: ...
+
+ @typing.final
+ class Unlimited(MaxTopicSize):
+ r"""
+ Do not cap the topic size; it may grow without bound.
+ """
+
+ __match_args__ = ()
+ def __new__(cls) -> MaxTopicSize.Unlimited: ...
+ def __len__(self) -> builtins.int: ...
+ def __getitem__(self, key: builtins.int, /) -> typing.Any: ...
+
+ ...
+
[email protected]
+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.
+ """
+
@typing.final
class Permissions:
r"""
@@ -1574,6 +1721,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:
@@ -1598,15 +1775,43 @@ 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.
+
+ Rebuilds the list from scratch on every access; cache the result
+ rather than reading this repeatedly in a loop.
+ """
@typing.final
class TopicPermissions:
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index d7f878885..0dd8ad144 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@ build-backend = "maturin"
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.8.1.dev3"
+version = "0.8.1.dev4"
description = "Apache Iggy is the persistent message streaming platform
written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of
processing millions of messages per second."
readme = "README.md"
license = { file = "LICENSE" }
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index a7f0cfd51..8d3aa1594 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -17,7 +17,8 @@
use bytes::Bytes;
use iggy::prelude::{
- Consumer as RustConsumer, IggyClient as RustIggyClient, IggyMessage as
RustMessage,
+ AutoCommit as RustAutoCommit, Consumer as RustConsumer, IggyClient as
RustIggyClient,
+ IggyExpiry as RustIggyExpiry, IggyMessage as RustMessage, MaxTopicSize as
RustMaxTopicSize,
PollingStrategy as RustPollingStrategy, *,
};
use pyo3::PyRef;
@@ -38,7 +39,7 @@ use crate::permissions::Permissions as PyPermissions;
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 crate::user::{
UserInfo as PyUserInfo, UserInfoDetails as PyUserInfoDetails, UserStatus
as PyUserStatus,
};
@@ -53,6 +54,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 {
@@ -409,7 +436,22 @@ impl IggyClient {
}
/// Creates a new topic with the given parameters.
- /// Returns Ok(()) on successful topic creation or a PyRuntimeError on
failure.
+ ///
+ /// 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)
)]
@@ -427,22 +469,15 @@ 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)
- .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?,
- None => CompressionAlgorithm::default(),
- };
-
- let expiry = match message_expiry {
- Some(delta) =>
IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)),
- None => IggyExpiry::ServerDefault,
- };
-
- let max_size = max_topic_size.map_or(MaxTopicSize::ServerDefault,
MaxTopicSize::from);
+ 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();
@@ -525,14 +560,15 @@ 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.
///
/// 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)
)]
@@ -550,22 +586,15 @@ 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)
- .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?,
- None => CompressionAlgorithm::default(),
- };
-
- let expiry = match message_expiry {
- Some(delta) =>
IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)),
- None => IggyExpiry::ServerDefault,
- };
-
- let max_size = max_topic_size.map_or(MaxTopicSize::ServerDefault,
MaxTopicSize::from);
+ 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)?;
@@ -1020,16 +1049,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, _>(
@@ -1045,7 +1074,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 0b6066b68..4d64fc626 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 = 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(),
+ ));
+ }
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/lib.rs b/foreign/python/src/lib.rs
index 9bb2308f4..4f552eadc 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -36,7 +36,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};
use user::{UserInfo, UserInfoDetails, UserStatus};
use user_headers::{HeaderKey, HeaderValue, UserHeaders};
@@ -49,6 +49,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 0c9028ef4..178f90dd5 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -15,9 +15,155 @@
// 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::{
+ 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_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};
+
+use crate::consumer::py_delta_to_iggy_duration;
+
+/// The expiry of the messages in a topic.
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+pub enum IggyExpiry {
+ /// Use the message expiry configured on the server for this topic,
+ /// rather than an explicit value set by the client.
+ ServerDefault(),
+ /// Expire messages this long after they are appended to the topic.
+ ///
+ /// `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(),
+}
+
+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())?,
+ },
+ RustIggyExpiry::NeverExpire => IggyExpiry::NeverExpire(),
+ })
+ }
+}
+
+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 } => {
+ 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,
+ })
+ }
+}
+
+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| {
+ PyDelta::new(
+ py,
+ days as i32,
+ secs_of_day as i32,
+ duration.subsec_micros() as i32,
+ true,
+ )
+ .map(|delta| delta.unbind())
+ .map_err(|err| {
+ PyValueError::new_err(format!(
+ "topic message expiry duration does not fit within timedelta
bounds: {err}"
+ ))
+ })
+ })
+}
+
+/// The maximum size of a topic.
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+pub enum MaxTopicSize {
+ /// Use the maximum topic size configured on the server, rather than an
+ /// explicit value set by the client.
+ ServerDefault(),
+ /// 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 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(),
+}
+
+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(),
+ }
+ }
+}
+
+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 } => {
+ 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,
+ })
+ }
+}
#[gen_stub_pyclass]
#[pyclass]
@@ -57,6 +203,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) -> PyResult<IggyExpiry> {
+ self.inner.message_expiry.try_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 +282,105 @@ 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) -> PyResult<IggyExpiry> {
+ self.inner.message_expiry.try_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.
+ ///
+ /// 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
+ .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
+ }
}
diff --git a/foreign/python/tests/test_topic.py
b/foreign/python/tests/test_topic.py
index 62bb37a5b..d76d16f35 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -19,7 +19,7 @@ from datetime import timedelta
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 @@ class TestCreateTopic:
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
@@ -229,15 +233,20 @@ class TestCreateTopic:
@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(microseconds=1)),
+ 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(
- 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()
@@ -254,13 +263,47 @@ class TestCreateTopic:
topic = await iggy_client.get_topic(stream_name, topic_name)
assert topic is not None
assert topic.name == topic_name
+ assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
+ assert topic.message_expiry.duration == message_expiry.duration
@pytest.mark.asyncio
- @pytest.mark.parametrize("invalid_message_expiry", [1, "1s", object()])
+ @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 an ExpireDuration at a reserved
boundary."""
+ 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(invalid_duration),
+ )
+
+ @pytest.mark.asyncio
+ @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()
@@ -276,15 +319,19 @@ class TestCreateTopic:
@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,
+ (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
+ self,
+ iggy_client: IggyClient,
+ unique_name,
+ max_topic_size: MaxTopicSize,
+ expected_kind: str,
):
"""Test create_topic accepts supported maximum topic size values."""
stream_name = unique_name()
@@ -301,21 +348,29 @@ class TestCreateTopic:
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 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),
(2e64, TypeError),
+ (0, ValueError),
+ (2**64 - 1, ValueError), # u64::MAX is reserved for Unlimited
],
)
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."""
@@ -329,7 +384,7 @@ class TestCreateTopic:
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
@@ -540,6 +595,28 @@ class TestGetTopic:
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."""
@@ -690,6 +767,44 @@ class TestGetTopics:
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
@@ -812,11 +927,13 @@ class TestUpdateTopic:
)
@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()
@@ -895,17 +1012,19 @@ class TestUpdateTopic:
@pytest.mark.asyncio
@pytest.mark.parametrize(
- ("max_topic_size", "expected_exception"),
+ ("max_topic_size_bytes", "expected_exception"),
[
(-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(
self,
iggy_client: IggyClient,
unique_name,
- max_topic_size,
+ max_topic_size_bytes,
expected_exception,
):
"""Test update_topic rejects invalid maximum topic size values."""
@@ -922,7 +1041,7 @@ class TestUpdateTopic:
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
@@ -942,19 +1061,63 @@ class TestUpdateTopic:
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)
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(
+ "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"),
+ [
+ (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
+ self,
+ iggy_client: IggyClient,
+ unique_name,
+ max_topic_size: MaxTopicSize,
+ expected_kind: str,
):
"""Test update_topic accepts supported maximum topic size values."""
stream_name = unique_name()
@@ -975,8 +1138,14 @@ class TestUpdateTopic:
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 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(
@@ -1213,12 +1382,18 @@ class TestPurgeTopic:
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(
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index 55960c278..0ae74df8d 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -12,7 +12,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.8.1.dev3"
+version = "0.8.1.dev4"
source = { editable = "." }
[package.optional-dependencies]
@@ -335,7 +335,7 @@ name = "exceptiongroup"
version = "1.3.1"
source = { registry = "https://pypi.org/simple"; }
dependencies = [
- { name = "typing-extensions" },
+ { name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url =
"https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz";,
hash =
"sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size
= 30371, upload-time = "2025-11-21T23:01:54.787Z" }
wheels = [
