From 476a70ac4c43ec6aea3be2bf52966d60f0f9a813 Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Thu, 16 Jul 2026 14:36:20 +0200 Subject: [PATCH 1/9] feat(python): expose structured IggyError --- foreign/python/apache_iggy.pyi | 15 +++ foreign/python/src/client.rs | 39 ++++---- foreign/python/src/consumer.rs | 24 ++--- foreign/python/src/error.rs | 63 ++++++++++++ foreign/python/src/identifier.rs | 15 ++- foreign/python/src/lib.rs | 3 + foreign/python/src/send_message.rs | 9 +- foreign/python/tests/test_connectivity.py | 112 ++++++++++++++++++---- 8 files changed, 220 insertions(+), 60 deletions(-) create mode 100644 foreign/python/src/error.rs diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 6a9ad26ba1..428c09826b 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -33,6 +33,7 @@ __all__ = [ "ConsumerGroupMember", "IggyClient", "IggyConsumer", + "IggyError", "PollingStrategy", "ReceiveMessage", "SendMessage", @@ -638,6 +639,20 @@ class IggyConsumer: Returns an awaitable that completes when shutdown is signaled or a PyRuntimeError on failure. """ +@typing.final +class IggyError(builtins.Exception): + @property + def code(self) -> builtins.int: ... + @property + def name(self) -> builtins.str: ... + @property + def message(self) -> builtins.str: ... + def __new__( + cls, code: builtins.int, name: builtins.str, message: builtins.str + ) -> IggyError: ... + def __str__(self) -> builtins.str: ... + def __repr__(self) -> builtins.str: ... + class PollingStrategy: @typing.final class Offset(PollingStrategy): diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index b604f62c6d..8de6782983 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -32,6 +32,7 @@ use crate::consumer::{ AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer, py_delta_to_iggy_duration, }; +use crate::error::IggyError as PyIggyError; use crate::identifier::PyIdentifier; use crate::receive_message::{PollingStrategy, ReceiveMessage}; use crate::send_message::SendMessage; @@ -63,7 +64,7 @@ impl IggyClient { .with_tcp() .with_server_address(conn.unwrap_or("127.0.0.1:8090".to_string())) .build() - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(IggyClient { inner: Arc::new(client), }) @@ -80,7 +81,7 @@ impl IggyClient { connection_string: String, ) -> PyResult { let client = RustIggyClient::from_connection_string(&connection_string) - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(Self { inner: Arc::new(client), }) @@ -96,7 +97,7 @@ impl IggyClient { inner .ping() .await - .map_err(|e| PyErr::new::(e.to_string())) + .map_err(|e| PyIggyError::new_err_from_rust(e)) }) } @@ -114,7 +115,7 @@ impl IggyClient { inner .login_user(&username, &password) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -128,7 +129,7 @@ impl IggyClient { inner .connect() .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -143,7 +144,7 @@ impl IggyClient { inner .create_stream(&name) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -163,7 +164,7 @@ impl IggyClient { let stream = inner .get_stream(&stream_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(stream.map(StreamDetails::from)) }) } @@ -219,7 +220,7 @@ impl IggyClient { max_size, ) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -241,7 +242,7 @@ impl IggyClient { let topic = inner .get_topic(&stream_id, &topic_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(topic.map(TopicDetails::from)) }) } @@ -269,7 +270,7 @@ impl IggyClient { let topics = inner .get_topics(&stream_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(topics.into_iter().map(Topic::from).collect::>()) }) } @@ -374,7 +375,7 @@ impl IggyClient { inner .delete_topic(&stream_id, &topic_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -405,7 +406,7 @@ impl IggyClient { inner .purge_topic(&stream_id, &topic_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -439,7 +440,7 @@ impl IggyClient { inner .create_consumer_group(&stream_id, &topic_id, &name) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -475,7 +476,7 @@ impl IggyClient { let group = inner .get_consumer_group(&stream_id, &topic_id, &group_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(group.map(PyConsumerGroupDetails::from)) }) } @@ -507,7 +508,7 @@ impl IggyClient { let groups = inner .get_consumer_groups(&stream_id, &topic_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(groups .into_iter() .map(PyConsumerGroup::from) @@ -547,7 +548,7 @@ impl IggyClient { inner .send_messages(&stream, &topic, &partitioning, messages.as_mut()) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -585,7 +586,7 @@ impl IggyClient { auto_commit, ) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; let messages = polled_messages .messages .into_iter() @@ -646,7 +647,7 @@ impl IggyClient { let mut builder = self .inner .consumer_group(name, stream, topic) - .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyIggyError::new_err_from_rust(e))? .without_encryptor() .partition(partition_id); @@ -704,7 +705,7 @@ impl IggyClient { consumer .init() .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(IggyConsumer { inner: Arc::new(Mutex::new(consumer)), }) diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 0b6066b686..2be3f21f7b 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs @@ -25,7 +25,7 @@ use iggy::prelude::{ AutoCommitWhen as RustAutoCommitWhen, ConsumerGroup as RustConsumerGroup, ConsumerGroupDetails as RustConsumerGroupDetails, ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyDuration, - IggyError, ReceivedMessage, + IggyError as RustIggyError, ReceivedMessage, }; use pyo3::exceptions::PyStopAsyncIteration; use pyo3::types::{PyDelta, PyDeltaAccess}; @@ -41,6 +41,7 @@ use tokio::task::JoinHandle; use crate::identifier::PyIdentifier; use crate::receive_message::ReceiveMessage; +use crate::error::IggyError as PyIggyError; /// A Python class representing the Iggy consumer. /// It wraps the RustIggyConsumer and provides asynchronous functionality @@ -110,7 +111,7 @@ impl IggyConsumer { .await .store_offset(offset, partition_id) .await - .map_err(|e| PyErr::new::(e.to_string())) + .map_err(|e| PyIggyError::new_err_from_rust(e)) }) } @@ -131,7 +132,7 @@ impl IggyConsumer { .await .delete_offset(partition_id) .await - .map_err(|e| PyErr::new::(e.to_string())) + .map_err(|e| PyIggyError::new_err_from_rust(e)) }) } @@ -167,7 +168,7 @@ impl IggyConsumer { let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); let task_locals = Python::attach(pyo3_async_runtimes::tokio::get_current_locals)?; - let handle_consume: JoinHandle>> = + let handle_consume: JoinHandle>> = get_runtime().spawn(scope(task_locals, async move { let task_locals = Python::attach(pyo3_async_runtimes::tokio::get_current_locals)?; @@ -212,7 +213,7 @@ impl IggyConsumer { consume_result .map_err(|e| PyErr::new::(e.to_string()))?? - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(|e| PyIggyError::new_err_from_rust(e))?; Ok(()) }) } @@ -365,9 +366,8 @@ impl ReceiveMessageIterator { inner: m.message, partition_id: m.partition_id, }) - .map_err(|e| { - PyErr::new::(e.to_string()) - })?) + .map_err(|e| PyIggyError::new_err_from_rust(e))? + ) } else { Err(PyStopAsyncIteration::new_err("No more messages")) } @@ -385,7 +385,7 @@ struct PyCallbackConsumer { } impl MessageConsumer for PyCallbackConsumer { - async fn consume(&self, received: ReceivedMessage) -> Result<(), IggyError> { + async fn consume(&self, received: ReceivedMessage) -> Result<(), RustIggyError> { let callback = self.callback.clone(); let task_locals = self.task_locals.clone().lock_owned().await; let task_locals = task_locals.clone(); @@ -402,10 +402,10 @@ impl MessageConsumer for PyCallbackConsumer { }) })) .await - .map_err(|_| IggyError::CannotReadMessage)? - .map_err(|_| IggyError::CannotReadMessage)? + .map_err(|_| RustIggyError::CannotReadMessage)? + .map_err(|_| RustIggyError::CannotReadMessage)? .await - .map_err(|_| IggyError::CannotReadMessage)?; + .map_err(|_| RustIggyError::CannotReadMessage)?; Ok(()) } } diff --git a/foreign/python/src/error.rs b/foreign/python/src/error.rs new file mode 100644 index 0000000000..042f66b482 --- /dev/null +++ b/foreign/python/src/error.rs @@ -0,0 +1,63 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use iggy::prelude::IggyError as RustIggyError; +use pyo3::exceptions::PyException; +use pyo3::prelude::*; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; + +#[gen_stub_pyclass] +#[pyclass(skip_from_py_object, extends=PyException)] +#[derive(Clone, Debug)] +pub struct IggyError { + #[pyo3(get)] + code: u32, + #[pyo3(get)] + name: String, + #[pyo3(get)] + message: String, +} + +#[gen_stub_pymethods] +#[pymethods] +impl IggyError { + #[new] + fn new(code: u32, name: String, message: String) -> Self { + Self { + code, + name, + message, + } + } + + fn __str__(&self) -> String { + self.message.clone() + } + + fn __repr__(&self) -> String { + format!( + "IggyError(code={}, name='{}', message='{}')", + self.code, self.name, self.message + ) + } +} + +impl IggyError { + pub fn new_err_from_rust(e: RustIggyError) -> PyErr { + PyErr::new::((e.as_code(), e.as_string().to_string(), e.to_string())) + } +} diff --git a/foreign/python/src/identifier.rs b/foreign/python/src/identifier.rs index a113e5e542..ad7cd548c0 100644 --- a/foreign/python/src/identifier.rs +++ b/foreign/python/src/identifier.rs @@ -18,12 +18,11 @@ use std::str::FromStr; use iggy::prelude::{IdKind, Identifier}; -use pyo3::{ - exceptions::{PyRuntimeError, PyValueError}, - prelude::*, -}; +use pyo3::prelude::*; use pyo3_stub_gen::impl_stub_type; +use crate::error::IggyError as PyIggyError; + #[derive(FromPyObject, IntoPyObject)] pub(crate) enum PyIdentifier { #[pyo3(transparent, annotation = "str")] @@ -39,10 +38,10 @@ impl TryFrom for Identifier { fn try_from(py_identifier: PyIdentifier) -> Result { match py_identifier { PyIdentifier::String(s) => { - Identifier::from_str(&s).map_err(|e| PyErr::new::(e.to_string())) + Identifier::from_str(&s).map_err(|e| PyIggyError::new_err_from_rust(e)) } PyIdentifier::Int(i) => { - Identifier::numeric(i).map_err(|e| PyErr::new::(e.to_string())) + Identifier::numeric(i).map_err(|e| PyIggyError::new_err_from_rust(e)) } } } @@ -56,11 +55,11 @@ impl TryFrom<&Identifier> for PyIdentifier { IdKind::String => val .get_string_value() .map(PyIdentifier::String) - .map_err(|e| PyErr::new::(e.to_string())), + .map_err(|e| PyIggyError::new_err_from_rust(e)), IdKind::Numeric => val .get_u32_value() .map(PyIdentifier::Int) - .map_err(|e| PyErr::new::(e.to_string())), + .map_err(|e| PyIggyError::new_err_from_rust(e)), } } } diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index 8c78456062..fea4e3d2a2 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -17,6 +17,7 @@ pub mod client; mod consumer; +mod error; mod identifier; mod receive_message; mod send_message; @@ -28,6 +29,7 @@ use consumer::{ AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator, }; +use error::IggyError; use pyo3::prelude::*; use receive_message::{PollingStrategy, ReceiveMessage}; use send_message::SendMessage; @@ -40,6 +42,7 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/foreign/python/src/send_message.rs b/foreign/python/src/send_message.rs index 021125d7e6..33b11e98aa 100644 --- a/foreign/python/src/send_message.rs +++ b/foreign/python/src/send_message.rs @@ -24,6 +24,8 @@ use pyo3_stub_gen::{ }; use std::str::FromStr; +use crate::error::IggyError as PyIggyError; + /// A Python class representing a message to be sent. /// This class wraps a Rust message meant for sending, facilitating /// the creation of such messages from Python and their subsequent use in Rust. @@ -63,14 +65,15 @@ impl SendMessage { #[new] pub fn new(py: Python, data: PyMessagePayload) -> PyResult { let inner = match data { - PyMessagePayload::String(data) => RustIggyMessage::from_str(&data) - .map_err(|e| PyErr::new::(e.to_string()))?, + PyMessagePayload::String(data) => { + RustIggyMessage::from_str(&data).map_err(|e| PyIggyError::new_err_from_rust(e))? + } PyMessagePayload::Bytes(data) => { let bytes = Bytes::from(data.extract::>(py)?); RustIggyMessage::builder() .payload(bytes) .build() - .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyIggyError::new_err_from_rust(e))? } }; Ok(Self { inner }) diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 69516d74c1..58363584a8 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -15,9 +15,10 @@ # specific language governing permissions and limitations # under the License. + import pytest -from apache_iggy import IggyClient +from apache_iggy import IggyClient, IggyError from .utils import get_server_config, wait_for_ping, wait_for_server @@ -51,46 +52,121 @@ async def test_valid_connection_string(self, connection_string: str): await wait_for_ping(client, timeout=5, interval=1) @pytest.mark.parametrize( - ("invalid_value", "expected_error"), + ( + "invalid_value", + "expected_error_name", + "expected_error_code", + "expected_error_message", + ), [ - ("", "Invalid connection string"), - ("bad address", "Invalid connection string"), - ("http://{host}:{port}", "Invalid connection string"), - ("tcp://iggy:iggy@{host}:{port}", "Invalid connection string"), - ("{host}:", "Invalid connection string"), - (":{port}", "Invalid connection string"), - ("{host}:not-a-port", "Invalid connection string"), - ("{host}:70000", "Invalid connection string"), - ("iggy+tcp://", "Invalid connection string"), - ("iggy+tcp://iggy:iggy@", "Invalid connection string"), + ("", "InvalidConnectionString", 8000, "Invalid connection string"), + ( + "bad address", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "http://{host}:{port}", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "tcp://iggy:iggy@{host}:{port}", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ("{host}:", "InvalidConnectionString", 8000, "Invalid connection string"), + (":{port}", "InvalidConnectionString", 8000, "Invalid connection string"), + ( + "{host}:not-a-port", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "{host}:70000", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "iggy+tcp://", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "iggy+tcp://iggy:iggy@", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), ( "iggy+tcp://iggy:iggy@{host}:not-a-port", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "iggy+tcp://iggy:iggy@{host}:-1", + "InvalidConnectionString", + 8000, + "Invalid connection string", + ), + ( + "iggy+tcp://iggy:iggy@{host}:70000", + "InvalidConnectionString", + 8000, "Invalid connection string", ), - ("iggy+tcp://iggy:iggy@{host}:-1", "Invalid connection string"), - ("iggy+tcp://iggy:iggy@{host}:70000", "Invalid connection string"), ( "iggy+tcp://iggy:bad:format@{host}:{port}", + "InvalidConnectionString", + 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value", + "InvalidConnectionString", + 8000, "Invalid connection string", ), - ("iggy+quic://iggy:iggy@127.0.0.1:8080", "Cannot create endpoint"), + ( + "iggy+quic://iggy:iggy@127.0.0.1:8080", + "CannotCreateEndpoint", + 305, + "Cannot create endpoint", + ), ], ) - def test_invalid_connection_string(self, invalid_value: str, expected_error: str): + def test_invalid_connection_string( + self, + invalid_value: str, + expected_error_name: str, + expected_error_code: int, + expected_error_message: str, + ): """Test malformed server addresses and connection strings are rejected.""" host, port = get_server_config() value = invalid_value.format(host=host, port=port) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as excinfo_new: IggyClient(value) - with pytest.raises(RuntimeError, match=expected_error): + assert excinfo_new.value.code == expected_error_code + assert excinfo_new.value.name == expected_error_name + assert excinfo_new.value.message == expected_error_message + + with pytest.raises(IggyError) as excinfo_from_connection_string: IggyClient.from_connection_string(value) + assert excinfo_from_connection_string.value.code == expected_error_code + assert excinfo_from_connection_string.value.name == expected_error_name + assert excinfo_from_connection_string.value.message == expected_error_message + @pytest.mark.asyncio async def test_repeated_connect_does_not_error(self): """Test calling connect twice on the same client succeeds.""" From 82760f7a620029e56ed7665b9398f3cc9a135efd Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:23:39 +0200 Subject: [PATCH 2/9] feat(python): expose structured IggyError - changed tests accordingly, minor fixes, updated README --- foreign/python/README.md | 3 +- foreign/python/apache_iggy.pyi | 253 +++++------------- foreign/python/tests/test_connectivity.py | 117 +++++--- foreign/python/tests/test_consumer_group.py | 87 ++++-- .../python/tests/test_message_operations.py | 26 +- foreign/python/tests/test_stream.py | 38 ++- foreign/python/tests/test_topic.py | 119 ++++++-- 7 files changed, 353 insertions(+), 290 deletions(-) diff --git a/foreign/python/README.md b/foreign/python/README.md index f01754f6dc..2e72c82bf7 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -48,7 +48,8 @@ cargo run --bin iggy-server -- --with-default-root-credentials --fresh # Using uv: uv sync --all-extras uv run maturin develop -uv run pytest tests/ -v # Run tests (requires iggy-server running) +uv run pytest tests/ -v # Run all tests (for release versions, requires iggy-server running) +uv run --no-sync pytest tests/ -v # Run tests without syncing (for development, implicit uv run's syncing overwrites the installations made by maturin develop) # Using pip: python3 -m venv .venv diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 428c09826b..3a80aa71c0 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -23,7 +23,6 @@ import builtins import collections.abc import datetime import typing - __all__ = [ "AutoCommit", "AutoCommitAfter", @@ -51,91 +50,75 @@ class AutoCommit: r""" The auto-commit is disabled and the offset must be stored manually by the consumer. """ - __match_args__ = () def __new__(cls) -> AutoCommit.Disabled: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class Interval(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server after a certain interval. """ - __match_args__ = ("_0",) @property def _0(self) -> datetime.timedelta: ... def __new__(cls, _0: datetime.timedelta) -> AutoCommit.Interval: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class IntervalOrWhen(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server after a certain interval or depending on the mode when consuming the messages. """ - - __match_args__ = ( - "_0", - "_1", - ) + __match_args__ = ("_0", "_1",) @property def _0(self) -> datetime.timedelta: ... @property def _1(self) -> AutoCommitWhen: ... - def __new__( - cls, _0: datetime.timedelta, _1: AutoCommitWhen - ) -> AutoCommit.IntervalOrWhen: ... + def __new__(cls, _0: datetime.timedelta, _1: AutoCommitWhen) -> AutoCommit.IntervalOrWhen: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class IntervalOrAfter(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server after a certain interval or depending on the mode after consuming the messages. """ - - __match_args__ = ( - "_0", - "_1", - ) + __match_args__ = ("_0", "_1",) @property def _0(self) -> datetime.timedelta: ... @property def _1(self) -> AutoCommitAfter: ... - def __new__( - cls, _0: datetime.timedelta, _1: AutoCommitAfter - ) -> AutoCommit.IntervalOrAfter: ... + def __new__(cls, _0: datetime.timedelta, _1: AutoCommitAfter) -> AutoCommit.IntervalOrAfter: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class When(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server depending on the mode when consuming the messages. """ - __match_args__ = ("_0",) @property def _0(self) -> AutoCommitWhen: ... def __new__(cls, _0: AutoCommitWhen) -> AutoCommit.When: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class After(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server depending on the mode after consuming the messages. """ - __match_args__ = ("_0",) @property def _0(self) -> AutoCommitAfter: ... def __new__(cls, _0: AutoCommitAfter) -> AutoCommit.After: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + ... class AutoCommitAfter: @@ -147,38 +130,33 @@ class AutoCommitAfter: r""" The offset is stored on the server after all the messages are consumed. """ - __match_args__ = () def __new__(cls) -> AutoCommitAfter.ConsumingAllMessages: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEachMessage(AutoCommitAfter): r""" The offset is stored on the server after consuming each message. """ - __match_args__ = () def __new__(cls) -> AutoCommitAfter.ConsumingEachMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEveryNthMessage(AutoCommitAfter): r""" The offset is stored on the server after consuming every Nth message. """ - __match_args__ = ("_0",) @property def _0(self) -> builtins.int: ... - def __new__( - cls, _0: builtins.int - ) -> AutoCommitAfter.ConsumingEveryNthMessage: ... + def __new__(cls, _0: builtins.int) -> AutoCommitAfter.ConsumingEveryNthMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + ... class AutoCommitWhen: @@ -190,49 +168,43 @@ class AutoCommitWhen: r""" The offset is stored on the server when the messages are received. """ - __match_args__ = () def __new__(cls) -> AutoCommitWhen.PollingMessages: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingAllMessages(AutoCommitWhen): r""" The offset is stored on the server when all the messages are consumed. """ - __match_args__ = () def __new__(cls) -> AutoCommitWhen.ConsumingAllMessages: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEachMessage(AutoCommitWhen): r""" The offset is stored on the server when consuming each message. """ - __match_args__ = () def __new__(cls) -> AutoCommitWhen.ConsumingEachMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEveryNthMessage(AutoCommitWhen): r""" The offset is stored on the server when consuming every Nth message. """ - __match_args__ = ("_0",) @property def _0(self) -> builtins.int: ... - def __new__( - cls, _0: builtins.int - ) -> AutoCommitWhen.ConsumingEveryNthMessage: ... + def __new__(cls, _0: builtins.int) -> AutoCommitWhen.ConsumingEveryNthMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + ... @typing.final @@ -329,9 +301,7 @@ class IggyClient: Returns `Ok(())` if the server responds successfully, or a `PyRuntimeError` if the connection fails. """ - def login_user( - self, username: builtins.str, password: builtins.str - ) -> collections.abc.Awaitable[None]: + def login_user(self, username: builtins.str, password: builtins.str) -> collections.abc.Awaitable[None]: r""" Logs in the user with the given credentials. Returns `Ok(())` on success, or a PyRuntimeError on failure. @@ -346,67 +316,41 @@ class IggyClient: Creates a new stream with the provided ID and name. Returns Ok(()) on successful stream creation or a PyRuntimeError on failure. """ - def get_stream( - self, stream_id: builtins.str | builtins.int - ) -> collections.abc.Awaitable[StreamDetails | None]: + def get_stream(self, stream_id: builtins.str | builtins.int) -> collections.abc.Awaitable[StreamDetails | None]: r""" Gets stream by id. Returns Option of stream details or a PyRuntimeError on failure. """ - def create_topic( - self, - stream: builtins.str | builtins.int, - name: builtins.str, - 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, - ) -> collections.abc.Awaitable[None]: + def create_topic(self, stream: builtins.str | builtins.int, name: builtins.str, 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) -> collections.abc.Awaitable[None]: r""" Creates a new topic with the given parameters. Returns Ok(()) on successful topic creation or a PyRuntimeError on failure. """ - def get_topic( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - ) -> collections.abc.Awaitable[TopicDetails | None]: + def get_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[TopicDetails | None]: r""" Gets topic by stream and id. Returns Option of topic details or a PyRuntimeError on failure. """ - def get_topics( - self, stream_id: builtins.str | builtins.int - ) -> collections.abc.Awaitable[list[Topic]]: + def get_topics(self, stream_id: builtins.str | builtins.int) -> collections.abc.Awaitable[list[Topic]]: r""" Get all topics in a stream. - + Args: stream_id: Stream identifier as `str | int`. - + Returns: An awaitable that resolves to `list[Topic]`. - + Raises: PyRuntimeError: If the identifier is invalid or the request fails. """ - def update_topic( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - 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, - ) -> collections.abc.Awaitable[None]: + def update_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int, 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) -> collections.abc.Awaitable[None]: r""" Update an existing topic. - + This is a full replacement: any optional parameter left unset is reset to its server default rather than preserved. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. @@ -415,152 +359,100 @@ class IggyClient: 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`. - + Returns: An awaitable that resolves to `None` when the topic is updated. - + Raises: PyRuntimeError: If an argument is invalid or the request fails. """ - def delete_topic( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - ) -> collections.abc.Awaitable[None]: + def delete_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[None]: r""" Delete a topic from a stream. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - + Returns: An awaitable that resolves to `None` when the topic is deleted. - + Raises: PyRuntimeError: If an identifier is invalid or the request fails. """ - def purge_topic( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - ) -> collections.abc.Awaitable[None]: + def purge_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[None]: r""" Purge all messages from a topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - + Returns: An awaitable that resolves to `None` when the topic is purged. - + Raises: PyRuntimeError: If an identifier is invalid or the request fails. """ - def create_consumer_group( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - name: builtins.str, - ) -> collections.abc.Awaitable[None]: + def create_consumer_group(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int, name: builtins.str) -> collections.abc.Awaitable[None]: r""" Create a consumer group for a stream and topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. name: Consumer group name as `str`. - + Returns: An awaitable that resolves to `None` when the consumer group is created. - + Raises: PyValueError: If an identifier is invalid. PyRuntimeError: If the request fails. """ - def get_consumer_group( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - group_id: builtins.str | builtins.int, - ) -> collections.abc.Awaitable[ConsumerGroupDetails | None]: + def get_consumer_group(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int, group_id: builtins.str | builtins.int) -> collections.abc.Awaitable[ConsumerGroupDetails | None]: r""" Retrieve details for a consumer group from the specified stream and topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. group_id: Consumer group identifier as `str | int`. - + Returns: An awaitable that resolves to `ConsumerGroupDetails` if the consumer group exists, or `None` otherwise. - + Raises: PyValueError: If an identifier is invalid. PyRuntimeError: If the request fails. """ - def get_consumer_groups( - self, - stream_id: builtins.str | builtins.int, - topic_id: builtins.str | builtins.int, - ) -> collections.abc.Awaitable[list[ConsumerGroup]]: + def get_consumer_groups(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[list[ConsumerGroup]]: r""" Get all consumer groups for the specified stream and topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - + Returns: An awaitable that resolves to `list[ConsumerGroup]`. - + Raises: PyValueError: If an identifier is invalid. PyRuntimeError: If the request fails. """ - def send_messages( - self, - stream: builtins.str | builtins.int, - topic: builtins.str | builtins.int, - partitioning: builtins.int, - messages: list[SendMessage], - ) -> collections.abc.Awaitable[None]: + def send_messages(self, stream: builtins.str | builtins.int, topic: builtins.str | builtins.int, partitioning: builtins.int, messages: list[SendMessage]) -> collections.abc.Awaitable[None]: r""" Sends a list of messages to the specified topic. Returns Ok(()) on successful sending or a PyRuntimeError on failure. """ - def poll_messages( - self, - stream: builtins.str | builtins.int, - topic: builtins.str | builtins.int, - partition_id: builtins.int, - polling_strategy: PollingStrategy, - count: builtins.int, - auto_commit: builtins.bool, - ) -> collections.abc.Awaitable[list[ReceiveMessage]]: + def poll_messages(self, stream: builtins.str | builtins.int, topic: builtins.str | builtins.int, partition_id: builtins.int, polling_strategy: PollingStrategy, count: builtins.int, auto_commit: builtins.bool) -> collections.abc.Awaitable[list[ReceiveMessage]]: r""" Polls for messages from the specified topic and partition. Returns a list of received messages or a PyRuntimeError on failure. """ - def consumer_group( - self, - name: builtins.str, - stream: builtins.str, - topic: builtins.str, - partition_id: builtins.int | None = None, - polling_strategy: PollingStrategy | None = None, - batch_length: builtins.int | None = None, - auto_commit: AutoCommit | None = None, - create_consumer_group_if_not_exists: builtins.bool = True, - auto_join_consumer_group: builtins.bool = True, - poll_interval: datetime.timedelta | None = None, - polling_retry_interval: datetime.timedelta | None = None, - init_retries: builtins.int | None = None, - init_retry_interval: datetime.timedelta | None = None, - allow_replay: builtins.bool = False, - ) -> collections.abc.Awaitable[IggyConsumer]: + def consumer_group(self, name: builtins.str, stream: builtins.str, topic: builtins.str, partition_id: builtins.int | None = None, polling_strategy: PollingStrategy | None = None, batch_length: builtins.int | None = None, auto_commit: AutoCommit | None = None, create_consumer_group_if_not_exists: builtins.bool = True, auto_join_consumer_group: builtins.bool = True, poll_interval: datetime.timedelta | None = None, polling_retry_interval: datetime.timedelta | None = None, init_retries: builtins.int | None = None, init_retry_interval: datetime.timedelta | None = None, allow_replay: builtins.bool = False) -> collections.abc.Awaitable[IggyConsumer]: r""" Creates a new consumer group consumer. Returns the consumer or a PyRuntimeError on failure. @@ -573,9 +465,7 @@ class IggyConsumer: It wraps the RustIggyConsumer and provides asynchronous functionality through the contained runtime. """ - def get_last_consumed_offset( - self, partition_id: builtins.int - ) -> builtins.int | None: + def get_last_consumed_offset(self, partition_id: builtins.int) -> builtins.int | None: r""" Get the last consumed offset or `None` if no offset has been consumed yet. """ @@ -599,18 +489,14 @@ class IggyConsumer: r""" Gets the name of the topic this consumer group is configured for. """ - def store_offset( - self, offset: builtins.int, partition_id: builtins.int | None - ) -> collections.abc.Awaitable[None]: + def store_offset(self, offset: builtins.int, partition_id: builtins.int | None) -> collections.abc.Awaitable[None]: r""" Stores the provided offset for the provided partition id or if none is specified uses the current partition id for the consumer group. Returns `Ok(())` if the server responds successfully, or a `PyRuntimeError` if the operation fails. """ - def delete_offset( - self, partition_id: builtins.int | None - ) -> collections.abc.Awaitable[None]: + def delete_offset(self, partition_id: builtins.int | None) -> collections.abc.Awaitable[None]: r""" Deletes the offset for the provided partition id or if none is specified uses the current partition id for the consumer group. @@ -627,13 +513,7 @@ class IggyConsumer: only the interval part is applied; the `after` mode is ignored. Use `consume_messages()` if you need commit-after-processing semantics. """ - def consume_messages( - self, - callback: collections.abc.Callable[ - [ReceiveMessage], collections.abc.Awaitable[None] - ], - shutdown_event: asyncio.Event | None, - ) -> collections.abc.Awaitable[None]: + def consume_messages(self, callback: collections.abc.Callable[[ReceiveMessage], collections.abc.Awaitable[None]], shutdown_event: asyncio.Event | None) -> collections.abc.Awaitable[None]: r""" Consumes messages continuously using a callback function and an optional `asyncio.Event` for signaling shutdown. Returns an awaitable that completes when shutdown is signaled or a PyRuntimeError on failure. @@ -647,9 +527,7 @@ class IggyError(builtins.Exception): def name(self) -> builtins.str: ... @property def message(self) -> builtins.str: ... - def __new__( - cls, code: builtins.int, name: builtins.str, message: builtins.str - ) -> IggyError: ... + def __new__(cls, code: builtins.int, name: builtins.str, message: builtins.str) -> IggyError: ... def __str__(self) -> builtins.str: ... def __repr__(self) -> builtins.str: ... @@ -660,29 +538,29 @@ class PollingStrategy: @property def value(self) -> builtins.int: ... def __new__(cls, value: builtins.int) -> PollingStrategy.Offset: ... - + @typing.final class Timestamp(PollingStrategy): __match_args__ = ("value",) @property def value(self) -> builtins.int: ... def __new__(cls, value: builtins.int) -> PollingStrategy.Timestamp: ... - + @typing.final class First(PollingStrategy): __match_args__ = () def __new__(cls) -> PollingStrategy.First: ... - + @typing.final class Last(PollingStrategy): __match_args__ = () def __new__(cls) -> PollingStrategy.Last: ... - + @typing.final class Next(PollingStrategy): __match_args__ = () def __new__(cls) -> PollingStrategy.Next: ... - + ... @typing.final @@ -806,3 +684,4 @@ class TopicDetails: r""" Replication factor for the topic. """ + diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 58363584a8..6eac8fae44 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -54,89 +54,89 @@ async def test_valid_connection_string(self, connection_string: str): @pytest.mark.parametrize( ( "invalid_value", - "expected_error_name", - "expected_error_code", - "expected_error_message", + "expected_iggy_error_name", + "expected_iggy_error_code", + "expected_iggy_error_message", ), [ - ("", "InvalidConnectionString", 8000, "Invalid connection string"), + ("", "invalid_connection_string", 8000, "Invalid connection string"), ( "bad address", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "http://{host}:{port}", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "tcp://iggy:iggy@{host}:{port}", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), - ("{host}:", "InvalidConnectionString", 8000, "Invalid connection string"), - (":{port}", "InvalidConnectionString", 8000, "Invalid connection string"), + ("{host}:", "invalid_connection_string", 8000, "Invalid connection string"), + (":{port}", "invalid_connection_string", 8000, "Invalid connection string"), ( "{host}:not-a-port", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "{host}:70000", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:not-a-port", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:-1", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:70000", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:bad:format@{host}:{port}", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value", - "InvalidConnectionString", + "invalid_connection_string", 8000, "Invalid connection string", ), ( "iggy+quic://iggy:iggy@127.0.0.1:8080", - "CannotCreateEndpoint", + "cannot_create_endpoint", 305, "Cannot create endpoint", ), @@ -145,27 +145,22 @@ async def test_valid_connection_string(self, connection_string: str): def test_invalid_connection_string( self, invalid_value: str, - expected_error_name: str, - expected_error_code: int, - expected_error_message: str, + expected_iggy_error_name: str, + expected_iggy_error_code: int, + expected_iggy_error_message: str, ): """Test malformed server addresses and connection strings are rejected.""" host, port = get_server_config() value = invalid_value.format(host=host, port=port) - with pytest.raises(IggyError) as excinfo_new: - IggyClient(value) - - assert excinfo_new.value.code == expected_error_code - assert excinfo_new.value.name == expected_error_name - assert excinfo_new.value.message == expected_error_message - with pytest.raises(IggyError) as excinfo_from_connection_string: IggyClient.from_connection_string(value) - assert excinfo_from_connection_string.value.code == expected_error_code - assert excinfo_from_connection_string.value.name == expected_error_name - assert excinfo_from_connection_string.value.message == expected_error_message + assert excinfo_from_connection_string.value.name == expected_iggy_error_name + assert excinfo_from_connection_string.value.code == expected_iggy_error_code + assert ( + excinfo_from_connection_string.value.message == expected_iggy_error_message + ) @pytest.mark.asyncio async def test_repeated_connect_does_not_error(self): @@ -188,19 +183,46 @@ async def test_login_before_connect_does_not_error(self): await wait_for_ping(client) @pytest.mark.parametrize( - ("username", "password", "expected_exception"), + ( + "username", + "password", + "expected_exception", + "expected_iggy_error_name", + "expected_iggy_error_code", + "expected_error_message", + ), [ - ("invalid-user", "iggy", RuntimeError), - ("iggy", "invalid-password", RuntimeError), - ("", "iggy", RuntimeError), - ("iggy", "", RuntimeError), - (None, "iggy", TypeError), - ("iggy", None, TypeError), + ( + "invalid-user", + "iggy", + IggyError, + "invalid_credentials", + 42, + "Invalid credentials", + ), + ( + "iggy", + "invalid-password", + IggyError, + "invalid_credentials", + 42, + "Invalid credentials", + ), + ("", "iggy", IggyError, "invalid_username", 43, "Invalid username"), + ("iggy", "", IggyError, "invalid_password", 44, "Invalid password"), + (None, "iggy", TypeError, None, None, "'None' is not an instance of 'str'"), + ("iggy", None, TypeError, None, None, "'None' is not an instance of 'str'"), ], ) @pytest.mark.asyncio async def test_login_with_invalid_credentials_fails( - self, username, password, expected_exception + self, + username: str, + password: str, + expected_exception: type[Exception], + expected_iggy_error_name: str, + expected_iggy_error_code: int, + expected_error_message: str, ): """Test login rejects invalid credentials and invalid argument values.""" host, port = get_server_config() @@ -210,10 +232,17 @@ async def test_login_with_invalid_credentials_fails( await client.connect() await wait_for_ping(client) - with pytest.raises(expected_exception): + with pytest.raises(expected_exception) as exc_info: result = client.login_user(username, password) await result + if expected_exception is IggyError: + assert exc_info.value.name == expected_iggy_error_name + assert exc_info.value.code == expected_iggy_error_code + assert exc_info.value.message == expected_error_message + else: + assert exc_info.value.args[0] == expected_error_message + @pytest.mark.asyncio async def test_relogin_with_invalid_credentials_fails(self): """Test re-login with invalid credentials fails after a successful login.""" @@ -225,9 +254,13 @@ async def test_relogin_with_invalid_credentials_fails(self): await wait_for_ping(client) await client.login_user("iggy", "iggy") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.login_user("iggy", "invalid-password") + assert exc_info.value.name == "invalid_credentials" + assert exc_info.value.code == 42 + assert exc_info.value.message == "Invalid credentials" + @pytest.mark.asyncio async def test_relogin_with_same_valid_credentials_does_not_error(self): """Test repeated login with the same valid credentials is allowed.""" diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py index d193a9a64c..e106d60b87 100644 --- a/foreign/python/tests/test_consumer_group.py +++ b/foreign/python/tests/test_consumer_group.py @@ -26,6 +26,7 @@ from apache_iggy import ( AutoCommit, IggyClient, + IggyError, PollingStrategy, ReceiveMessage, ) @@ -132,13 +133,17 @@ async def test_duplicate_consumer_group_creation_fails( group_name, ) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.create_consumer_group( stream_name, topic_name, group_name, ) + assert exc_info.value.name == "consumer_group_name_already_exists" + assert exc_info.value.code == 5004 + assert exc_info.value.message.startswith("Consumer group with name:") + @pytest.mark.asyncio async def test_create_consumer_group_requires_connection_and_auth( self, unique_name @@ -148,21 +153,29 @@ async def test_create_consumer_group_requires_connection_and_auth( wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.create_consumer_group( unique_name(), unique_name(), unique_name(), ) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.create_consumer_group( unique_name(), unique_name(), unique_name(), ) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestGetConsumerGroup: """Test consumer group retrieval via get_consumer_group.""" @@ -288,21 +301,29 @@ async def test_get_consumer_group_requires_connection_and_auth(self, unique_name wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.get_consumer_group( unique_name(), unique_name(), unique_name(), ) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.get_consumer_group( unique_name(), unique_name(), unique_name(), ) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestGetConsumerGroups: """Test listing consumer groups via get_consumer_groups.""" @@ -428,13 +449,21 @@ async def test_get_consumer_groups_requires_connection_and_auth(self, unique_nam wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.get_consumer_groups(unique_name(), unique_name()) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.get_consumer_groups(unique_name(), unique_name()) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestConsumerGroup: """Test consumer group consumers.""" @@ -652,11 +681,15 @@ async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name poll_interval=timedelta(milliseconds=25), ) - with pytest.raises( - RuntimeError, - match=r"Consumer offset for consumer with ID: 0 was not found\.", - ): + with pytest.raises(IggyError) as exc_info: await consumer.delete_offset(partition_id) + + assert exc_info.value.name == "consumer_offset_not_found" + assert exc_info.value.code == 3021 + assert ( + exc_info.value.message + == "Consumer offset for consumer with ID: 0 was not found." + ) assert consumer.get_last_stored_offset(partition_id) is None await iggy_client.send_messages( @@ -1547,10 +1580,10 @@ async def test_consumer_group_can_disable_auto_create_when_group_exists( @pytest.mark.asyncio @pytest.mark.parametrize( - ("create_stream", "expected_error"), + ("create_stream", "expected_iggy_error_name", "expected_iggy_error_code"), [ - (False, "Stream with name:"), - (True, "Topic with name:"), + (False, "stream_name_not_found", 1010), + (True, "topic_name_not_found", 2011), ], ) async def test_consumer_group_missing_stream_or_topic_fails( @@ -1558,7 +1591,8 @@ async def test_consumer_group_missing_stream_or_topic_fails( iggy_client: IggyClient, unique_name, create_stream: bool, - expected_error: str, + expected_iggy_error_name: str, + expected_iggy_error_code: int, ): """Test initialization fails when the target stream or topic is missing.""" consumer_name = unique_name() @@ -1568,7 +1602,7 @@ async def test_consumer_group_missing_stream_or_topic_fails( if create_stream: await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError, match=expected_error): + with pytest.raises(IggyError) as exc_info: await iggy_client.consumer_group( consumer_name, stream_name, @@ -1580,6 +1614,9 @@ async def test_consumer_group_missing_stream_or_topic_fails( poll_interval=timedelta(milliseconds=25), ) + assert exc_info.value.name == expected_iggy_error_name + assert exc_info.value.code == expected_iggy_error_code + @pytest.mark.asyncio async def test_consumer_group_auto_create_disabled_fails_for_missing_group( self, iggy_client: IggyClient, unique_name @@ -1595,7 +1632,7 @@ async def test_consumer_group_auto_create_disabled_fails_for_missing_group( partitions_count=1, ) - with pytest.raises(RuntimeError, match="Consumer group with name:"): + with pytest.raises(IggyError) as exc_info: await iggy_client.consumer_group( unique_name(), stream_name, @@ -1608,6 +1645,10 @@ async def test_consumer_group_auto_create_disabled_fails_for_missing_group( poll_interval=timedelta(milliseconds=25), ) + assert exc_info.value.name == "consumer_group_name_not_found" + assert exc_info.value.code == 5003 + assert exc_info.value.message.startswith("Consumer group with name:") + @pytest.mark.asyncio @pytest.mark.parametrize( ("kwargs", "expected_error"), @@ -1655,7 +1696,7 @@ async def test_consumer_group_before_connect_fails(self, unique_name): host, port = get_server_config() client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.consumer_group( unique_name(), unique_name(), @@ -1667,6 +1708,10 @@ async def test_consumer_group_before_connect_fails(self, unique_name): poll_interval=timedelta(milliseconds=25), ) + assert exc_info.value.name == "disconnected" + assert exc_info.value.code == 8 + assert exc_info.value.message == "Disconnected" + @pytest.mark.asyncio async def test_consumer_group_before_login_fails(self, unique_name): """Test consumer_group requires authentication.""" @@ -1677,7 +1722,7 @@ async def test_consumer_group_before_login_fails(self, unique_name): await client.connect() await wait_for_ping(client) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.consumer_group( unique_name(), unique_name(), @@ -1689,6 +1734,10 @@ async def test_consumer_group_before_login_fails(self, unique_name): poll_interval=timedelta(milliseconds=25), ) + assert exc_info.value.name == "unauthenticated" + assert exc_info.value.code == 40 + assert exc_info.value.message == "Unauthenticated" + @pytest.mark.asyncio async def test_shutdown(self, iggy_client: IggyClient, unique_name): """Test that the consumer group can be signaled to shutdown.""" diff --git a/foreign/python/tests/test_message_operations.py b/foreign/python/tests/test_message_operations.py index da57637c05..cf02bf755e 100644 --- a/foreign/python/tests/test_message_operations.py +++ b/foreign/python/tests/test_message_operations.py @@ -20,7 +20,7 @@ import pytest -from apache_iggy import IggyClient, PollingStrategy +from apache_iggy import IggyClient, IggyError, PollingStrategy from apache_iggy import SendMessage as Message @@ -146,9 +146,13 @@ async def test_message_properties(self, iggy_client: IggyClient, unique_name): ) async def test_empty_payload_is_rejected(self, payload): """Test empty string and bytes payloads are rejected.""" - with pytest.raises(ValueError, match="Invalid message payload length"): + with pytest.raises(IggyError) as exc_info: Message(payload) + assert exc_info.value.name == "invalid_message_payload_length" + assert exc_info.value.code == 4025 + assert exc_info.value.message == "Invalid message payload length" + @pytest.mark.asyncio @pytest.mark.parametrize( "payload", @@ -361,7 +365,7 @@ async def test_poll_messages_with_count_zero_is_rejected( messages=[Message(message) for message in test_messages], ) - with pytest.raises(RuntimeError, match="Invalid messages count"): + with pytest.raises(IggyError) as exc_info: await iggy_client.poll_messages( stream=stream_name, topic=topic_name, @@ -371,6 +375,10 @@ async def test_poll_messages_with_count_zero_is_rejected( auto_commit=False, ) + assert exc_info.value.name == "invalid_messages_count" + assert exc_info.value.code == 4009 + assert exc_info.value.message == "Invalid messages count" + @pytest.mark.asyncio async def test_poll_messages_with_invalid_partition_id_raises( self, iggy_client: IggyClient, unique_name @@ -392,13 +400,7 @@ async def test_poll_messages_with_invalid_partition_id_raises( messages=[Message(f"Partition test - {unique_name()}")], ) - with pytest.raises( - RuntimeError, - match=( - r"Partition with ID: 0 for topic with ID: 0 " - r"for stream with ID: 0 was not found\." - ), - ): + with pytest.raises(IggyError) as exc_info: await iggy_client.poll_messages( stream=stream_name, topic=topic_name, @@ -408,6 +410,10 @@ async def test_poll_messages_with_invalid_partition_id_raises( auto_commit=False, ) + assert exc_info.value.name == "partition_not_found" + assert exc_info.value.code == 3007 + assert exc_info.value.message.startswith("Partition with ID:") + @pytest.mark.asyncio async def test_polling_strategy_last_with_count_one_returns_last_message( self, iggy_client: IggyClient, unique_name diff --git a/foreign/python/tests/test_stream.py b/foreign/python/tests/test_stream.py index 5e71a543a4..64595fc2c9 100644 --- a/foreign/python/tests/test_stream.py +++ b/foreign/python/tests/test_stream.py @@ -17,7 +17,7 @@ import pytest -from apache_iggy import IggyClient +from apache_iggy import IggyClient, IggyError from .utils import get_server_config, wait_for_ping, wait_for_server @@ -91,9 +91,13 @@ async def test_create_stream_invalid_names( """Test create_stream enforces byte-length validation.""" stream_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.create_stream(stream_name) + assert exc_info.value.name == "invalid_format" + assert exc_info.value.code == 4 + assert exc_info.value.message == "Invalid format" + @pytest.mark.asyncio async def test_get_stream_by_name_and_id( self, iggy_client: IggyClient, unique_name @@ -146,10 +150,12 @@ async def test_duplicate_stream_creation( await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError) as exc_info: + with pytest.raises(IggyError) as exc_info: await iggy_client.create_stream(stream_name) - assert "already exists" in str(exc_info.value) + assert exc_info.value.name == "stream_name_already_exists" + assert exc_info.value.code == 1012 + assert exc_info.value.message.startswith("Stream with name:") @pytest.mark.asyncio async def test_get_nonexistent_stream(self, iggy_client: IggyClient, unique_name): @@ -168,9 +174,13 @@ async def test_get_stream_before_connect_fails(self, unique_name): host, port = get_server_config() client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.get_stream(unique_name()) + assert exc_info.value.name == "disconnected" + assert exc_info.value.code == 8 + assert exc_info.value.message == "Disconnected" + @pytest.mark.asyncio async def test_get_stream_before_login_fails(self, unique_name): """Test get_stream requires authentication.""" @@ -180,18 +190,26 @@ async def test_get_stream_before_login_fails(self, unique_name): client = IggyClient(f"{host}:{port}") await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.get_stream(unique_name()) + assert exc_info.value.name == "unauthenticated" + assert exc_info.value.code == 40 + assert exc_info.value.message == "Unauthenticated" + @pytest.mark.asyncio async def test_create_stream_before_connect_fails(self, unique_name): """Test create_stream requires an established connection.""" host, port = get_server_config() client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.create_stream(unique_name()) + assert exc_info.value.name == "disconnected" + assert exc_info.value.code == 8 + assert exc_info.value.message == "Disconnected" + @pytest.mark.asyncio async def test_create_stream_before_login_fails(self, unique_name): """Test create_stream requires authentication.""" @@ -201,5 +219,9 @@ async def test_create_stream_before_login_fails(self, unique_name): client = IggyClient(f"{host}:{port}") await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await client.create_stream(unique_name()) + + assert exc_info.value.name == "unauthenticated" + assert exc_info.value.code == 40 + assert exc_info.value.message == "Unauthenticated" diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py index f3acc53954..5652733f53 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, IggyError, SendMessage from .utils import get_server_config, wait_for_ping, wait_for_server @@ -104,11 +104,15 @@ async def test_create_topic_invalid_names( await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.create_topic( stream=stream_name, name=topic_name, partitions_count=1 ) + assert exc_info.value.name == "invalid_format" + assert exc_info.value.code == 4 + assert exc_info.value.message == "Invalid format" + @pytest.mark.asyncio async def test_create_and_get_topic_with_numeric_stream_id( self, iggy_client: IggyClient, unique_name @@ -146,12 +150,14 @@ async def test_duplicate_topic_creation(self, iggy_client: IggyClient, unique_na stream=stream_name, name=topic_name, partitions_count=1 ) - with pytest.raises(RuntimeError) as exc_info: + with pytest.raises(IggyError) as exc_info: await iggy_client.create_topic( stream=stream_name, name=topic_name, partitions_count=1 ) - assert "already exists" in str(exc_info.value) + assert exc_info.value.name == "topic_name_already_exists" + assert exc_info.value.code == 2013 + assert exc_info.value.message.startswith("Topic with name:") @pytest.mark.asyncio async def test_topic_names_can_repeat_across_different_streams( @@ -306,7 +312,7 @@ async def test_create_topic_with_valid_max_topic_size( @pytest.mark.parametrize( ("max_topic_size", "expected_exception"), [ - (4563, RuntimeError), + (4563, IggyError), (-1, OverflowError), (2e64, TypeError), ], @@ -324,7 +330,7 @@ async def test_create_topic_invalid_max_topic_size( await iggy_client.create_stream(stream_name) - with pytest.raises(expected_exception): + with pytest.raises(expected_exception) as exc_info: await iggy_client.create_topic( stream=stream_name, name=topic_name, @@ -332,6 +338,11 @@ async def test_create_topic_invalid_max_topic_size( max_topic_size=max_topic_size, ) + if expected_exception is IggyError: + assert exc_info.value.name == "invalid_topic_size" + assert exc_info.value.code == 1019 + assert exc_info.value.message.startswith("Max topic size cannot") + @pytest.mark.asyncio @pytest.mark.parametrize( "replication_factor", @@ -403,12 +414,14 @@ async def test_create_topic_invalid_partitions_count( await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError) as exc_info: + with pytest.raises(IggyError) as exc_info: await iggy_client.create_topic( stream=stream_name, name=topic_name, partitions_count=partitions_count ) - assert "Too many partitions" in str(exc_info.value) + assert exc_info.value.name == "too_many_partitions" + assert exc_info.value.code == 2015 + assert exc_info.value.message == "Too many partitions" @pytest.mark.asyncio @pytest.mark.parametrize("partitions_count", [0, 1, 1000]) @@ -489,11 +502,15 @@ async def test_create_topic_in_nonexistent_stream( nonexistent_stream = unique_name() topic_name = unique_name() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.create_topic( stream=nonexistent_stream, name=topic_name, partitions_count=1 ) + assert exc_info.value.name == "stream_id_not_found" + assert exc_info.value.code == 1009 + assert exc_info.value.message.startswith("Stream with ID:") + @pytest.mark.asyncio async def test_create_topic_requires_connection_and_auth(self, unique_name): """Test create_topic fails both before connecting and before logging in.""" @@ -501,17 +518,25 @@ async def test_create_topic_requires_connection_and_auth(self, unique_name): wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.create_topic( stream=unique_name(), name=unique_name(), partitions_count=1 ) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.create_topic( stream=unique_name(), name=unique_name(), partitions_count=1 ) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestGetTopic: """Test topic retrieval via get_topic.""" @@ -572,13 +597,21 @@ async def test_get_topic_requires_connection_and_auth(self, unique_name): wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.get_topic(unique_name(), unique_name()) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.get_topic(unique_name(), unique_name()) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestGetTopics: """Test listing topics in a stream via get_topics.""" @@ -679,13 +712,21 @@ async def test_get_topics_requires_connection_and_auth(self, unique_name): wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.get_topics(unique_name()) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.get_topics(unique_name()) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestUpdateTopic: """Test updating topics via update_topic.""" @@ -1150,9 +1191,13 @@ async def test_delete_nonexistent_topic_fails( await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.delete_topic(stream_name, unique_name()) + assert exc_info.value.name == "topic_id_not_found" + assert exc_info.value.code == 2010 + assert exc_info.value.message.startswith("Topic with ID:") + @pytest.mark.asyncio async def test_delete_topic_twice_fails_second_time( self, iggy_client: IggyClient, unique_name @@ -1167,9 +1212,13 @@ async def test_delete_topic_twice_fails_second_time( ) await iggy_client.delete_topic(stream_name, topic_name) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.delete_topic(stream_name, topic_name) + assert exc_info.value.name == "topic_id_not_found" + assert exc_info.value.code == 2010 + assert exc_info.value.message.startswith("Topic with ID:") + @pytest.mark.asyncio async def test_delete_topic_requires_connection_and_auth(self, unique_name): """Test delete_topic fails both before connecting and before logging in.""" @@ -1177,13 +1226,21 @@ async def test_delete_topic_requires_connection_and_auth(self, unique_name): wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.delete_topic(unique_name(), unique_name()) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.delete_topic(unique_name(), unique_name()) + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" + class TestPurgeTopic: """Test purging topic messages via purge_topic.""" @@ -1271,17 +1328,25 @@ async def test_purge_nonexistent_topic_fails( await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.purge_topic(stream_name, unique_name()) + assert exc_info.value.name == "topic_id_not_found" + assert exc_info.value.code == 2010 + assert exc_info.value.message.startswith("Topic with ID:") + @pytest.mark.asyncio async def test_purge_topic_in_nonexistent_stream_fails( self, iggy_client: IggyClient, unique_name ): """Test purge_topic raises when the stream does not exist.""" - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.purge_topic(unique_name(), unique_name()) + assert exc_info.value.name == "stream_id_not_found" + assert exc_info.value.code == 1009 + assert exc_info.value.message.startswith("Stream with ID:") + @pytest.mark.asyncio async def test_purge_topic_requires_connection_and_auth(self, unique_name): """Test purge_topic fails both before connecting and before logging in.""" @@ -1289,9 +1354,17 @@ async def test_purge_topic_requires_connection_and_auth(self, unique_name): wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_disconnected: await client.purge_topic(unique_name(), unique_name()) + assert exc_info_disconnected.value.name == "disconnected" + assert exc_info_disconnected.value.code == 8 + assert exc_info_disconnected.value.message == "Disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_unauthenticated: await client.purge_topic(unique_name(), unique_name()) + + assert exc_info_unauthenticated.value.name == "unauthenticated" + assert exc_info_unauthenticated.value.code == 40 + assert exc_info_unauthenticated.value.message == "Unauthenticated" From 30d8eb45218e990265022bbf523a1137b9e2db46 Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Fri, 17 Jul 2026 14:55:03 +0200 Subject: [PATCH 3/9] feat(python): expose structured IggyError - minor doc and lint updates --- foreign/python/apache_iggy.pyi | 253 ++++++++++++++++------ foreign/python/src/error.rs | 2 + foreign/python/tests/test_connectivity.py | 2 +- 3 files changed, 190 insertions(+), 67 deletions(-) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 3a80aa71c0..428c09826b 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -23,6 +23,7 @@ import builtins import collections.abc import datetime import typing + __all__ = [ "AutoCommit", "AutoCommitAfter", @@ -50,75 +51,91 @@ class AutoCommit: r""" The auto-commit is disabled and the offset must be stored manually by the consumer. """ + __match_args__ = () def __new__(cls) -> AutoCommit.Disabled: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class Interval(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server after a certain interval. """ + __match_args__ = ("_0",) @property def _0(self) -> datetime.timedelta: ... def __new__(cls, _0: datetime.timedelta) -> AutoCommit.Interval: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class IntervalOrWhen(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server after a certain interval or depending on the mode when consuming the messages. """ - __match_args__ = ("_0", "_1",) + + __match_args__ = ( + "_0", + "_1", + ) @property def _0(self) -> datetime.timedelta: ... @property def _1(self) -> AutoCommitWhen: ... - def __new__(cls, _0: datetime.timedelta, _1: AutoCommitWhen) -> AutoCommit.IntervalOrWhen: ... + def __new__( + cls, _0: datetime.timedelta, _1: AutoCommitWhen + ) -> AutoCommit.IntervalOrWhen: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class IntervalOrAfter(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server after a certain interval or depending on the mode after consuming the messages. """ - __match_args__ = ("_0", "_1",) + + __match_args__ = ( + "_0", + "_1", + ) @property def _0(self) -> datetime.timedelta: ... @property def _1(self) -> AutoCommitAfter: ... - def __new__(cls, _0: datetime.timedelta, _1: AutoCommitAfter) -> AutoCommit.IntervalOrAfter: ... + def __new__( + cls, _0: datetime.timedelta, _1: AutoCommitAfter + ) -> AutoCommit.IntervalOrAfter: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class When(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server depending on the mode when consuming the messages. """ + __match_args__ = ("_0",) @property def _0(self) -> AutoCommitWhen: ... def __new__(cls, _0: AutoCommitWhen) -> AutoCommit.When: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class After(AutoCommit): r""" The auto-commit is enabled and the offset is stored on the server depending on the mode after consuming the messages. """ + __match_args__ = ("_0",) @property def _0(self) -> AutoCommitAfter: ... def __new__(cls, _0: AutoCommitAfter) -> AutoCommit.After: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + ... class AutoCommitAfter: @@ -130,33 +147,38 @@ class AutoCommitAfter: r""" The offset is stored on the server after all the messages are consumed. """ + __match_args__ = () def __new__(cls) -> AutoCommitAfter.ConsumingAllMessages: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEachMessage(AutoCommitAfter): r""" The offset is stored on the server after consuming each message. """ + __match_args__ = () def __new__(cls) -> AutoCommitAfter.ConsumingEachMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEveryNthMessage(AutoCommitAfter): r""" The offset is stored on the server after consuming every Nth message. """ + __match_args__ = ("_0",) @property def _0(self) -> builtins.int: ... - def __new__(cls, _0: builtins.int) -> AutoCommitAfter.ConsumingEveryNthMessage: ... + def __new__( + cls, _0: builtins.int + ) -> AutoCommitAfter.ConsumingEveryNthMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + ... class AutoCommitWhen: @@ -168,43 +190,49 @@ class AutoCommitWhen: r""" The offset is stored on the server when the messages are received. """ + __match_args__ = () def __new__(cls) -> AutoCommitWhen.PollingMessages: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingAllMessages(AutoCommitWhen): r""" The offset is stored on the server when all the messages are consumed. """ + __match_args__ = () def __new__(cls) -> AutoCommitWhen.ConsumingAllMessages: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEachMessage(AutoCommitWhen): r""" The offset is stored on the server when consuming each message. """ + __match_args__ = () def __new__(cls) -> AutoCommitWhen.ConsumingEachMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + @typing.final class ConsumingEveryNthMessage(AutoCommitWhen): r""" The offset is stored on the server when consuming every Nth message. """ + __match_args__ = ("_0",) @property def _0(self) -> builtins.int: ... - def __new__(cls, _0: builtins.int) -> AutoCommitWhen.ConsumingEveryNthMessage: ... + def __new__( + cls, _0: builtins.int + ) -> AutoCommitWhen.ConsumingEveryNthMessage: ... def __len__(self) -> builtins.int: ... def __getitem__(self, key: builtins.int, /) -> typing.Any: ... - + ... @typing.final @@ -301,7 +329,9 @@ class IggyClient: Returns `Ok(())` if the server responds successfully, or a `PyRuntimeError` if the connection fails. """ - def login_user(self, username: builtins.str, password: builtins.str) -> collections.abc.Awaitable[None]: + def login_user( + self, username: builtins.str, password: builtins.str + ) -> collections.abc.Awaitable[None]: r""" Logs in the user with the given credentials. Returns `Ok(())` on success, or a PyRuntimeError on failure. @@ -316,41 +346,67 @@ class IggyClient: Creates a new stream with the provided ID and name. Returns Ok(()) on successful stream creation or a PyRuntimeError on failure. """ - def get_stream(self, stream_id: builtins.str | builtins.int) -> collections.abc.Awaitable[StreamDetails | None]: + def get_stream( + self, stream_id: builtins.str | builtins.int + ) -> collections.abc.Awaitable[StreamDetails | None]: r""" Gets stream by id. Returns Option of stream details or a PyRuntimeError on failure. """ - def create_topic(self, stream: builtins.str | builtins.int, name: builtins.str, 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) -> collections.abc.Awaitable[None]: + def create_topic( + self, + stream: builtins.str | builtins.int, + name: builtins.str, + 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, + ) -> collections.abc.Awaitable[None]: r""" Creates a new topic with the given parameters. Returns Ok(()) on successful topic creation or a PyRuntimeError on failure. """ - def get_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[TopicDetails | None]: + def get_topic( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + ) -> collections.abc.Awaitable[TopicDetails | None]: r""" Gets topic by stream and id. Returns Option of topic details or a PyRuntimeError on failure. """ - def get_topics(self, stream_id: builtins.str | builtins.int) -> collections.abc.Awaitable[list[Topic]]: + def get_topics( + self, stream_id: builtins.str | builtins.int + ) -> collections.abc.Awaitable[list[Topic]]: r""" Get all topics in a stream. - + Args: stream_id: Stream identifier as `str | int`. - + Returns: An awaitable that resolves to `list[Topic]`. - + Raises: PyRuntimeError: If the identifier is invalid or the request fails. """ - def update_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int, 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) -> collections.abc.Awaitable[None]: + def update_topic( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + 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, + ) -> collections.abc.Awaitable[None]: r""" Update an existing topic. - + This is a full replacement: any optional parameter left unset is reset to its server default rather than preserved. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. @@ -359,100 +415,152 @@ class IggyClient: 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`. - + Returns: An awaitable that resolves to `None` when the topic is updated. - + Raises: PyRuntimeError: If an argument is invalid or the request fails. """ - def delete_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[None]: + def delete_topic( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + ) -> collections.abc.Awaitable[None]: r""" Delete a topic from a stream. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - + Returns: An awaitable that resolves to `None` when the topic is deleted. - + Raises: PyRuntimeError: If an identifier is invalid or the request fails. """ - def purge_topic(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[None]: + def purge_topic( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + ) -> collections.abc.Awaitable[None]: r""" Purge all messages from a topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - + Returns: An awaitable that resolves to `None` when the topic is purged. - + Raises: PyRuntimeError: If an identifier is invalid or the request fails. """ - def create_consumer_group(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int, name: builtins.str) -> collections.abc.Awaitable[None]: + def create_consumer_group( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + name: builtins.str, + ) -> collections.abc.Awaitable[None]: r""" Create a consumer group for a stream and topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. name: Consumer group name as `str`. - + Returns: An awaitable that resolves to `None` when the consumer group is created. - + Raises: PyValueError: If an identifier is invalid. PyRuntimeError: If the request fails. """ - def get_consumer_group(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int, group_id: builtins.str | builtins.int) -> collections.abc.Awaitable[ConsumerGroupDetails | None]: + def get_consumer_group( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + group_id: builtins.str | builtins.int, + ) -> collections.abc.Awaitable[ConsumerGroupDetails | None]: r""" Retrieve details for a consumer group from the specified stream and topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. group_id: Consumer group identifier as `str | int`. - + Returns: An awaitable that resolves to `ConsumerGroupDetails` if the consumer group exists, or `None` otherwise. - + Raises: PyValueError: If an identifier is invalid. PyRuntimeError: If the request fails. """ - def get_consumer_groups(self, stream_id: builtins.str | builtins.int, topic_id: builtins.str | builtins.int) -> collections.abc.Awaitable[list[ConsumerGroup]]: + def get_consumer_groups( + self, + stream_id: builtins.str | builtins.int, + topic_id: builtins.str | builtins.int, + ) -> collections.abc.Awaitable[list[ConsumerGroup]]: r""" Get all consumer groups for the specified stream and topic. - + Args: stream_id: Stream identifier as `str | int`. topic_id: Topic identifier as `str | int`. - + Returns: An awaitable that resolves to `list[ConsumerGroup]`. - + Raises: PyValueError: If an identifier is invalid. PyRuntimeError: If the request fails. """ - def send_messages(self, stream: builtins.str | builtins.int, topic: builtins.str | builtins.int, partitioning: builtins.int, messages: list[SendMessage]) -> collections.abc.Awaitable[None]: + def send_messages( + self, + stream: builtins.str | builtins.int, + topic: builtins.str | builtins.int, + partitioning: builtins.int, + messages: list[SendMessage], + ) -> collections.abc.Awaitable[None]: r""" Sends a list of messages to the specified topic. Returns Ok(()) on successful sending or a PyRuntimeError on failure. """ - def poll_messages(self, stream: builtins.str | builtins.int, topic: builtins.str | builtins.int, partition_id: builtins.int, polling_strategy: PollingStrategy, count: builtins.int, auto_commit: builtins.bool) -> collections.abc.Awaitable[list[ReceiveMessage]]: + def poll_messages( + self, + stream: builtins.str | builtins.int, + topic: builtins.str | builtins.int, + partition_id: builtins.int, + polling_strategy: PollingStrategy, + count: builtins.int, + auto_commit: builtins.bool, + ) -> collections.abc.Awaitable[list[ReceiveMessage]]: r""" Polls for messages from the specified topic and partition. Returns a list of received messages or a PyRuntimeError on failure. """ - def consumer_group(self, name: builtins.str, stream: builtins.str, topic: builtins.str, partition_id: builtins.int | None = None, polling_strategy: PollingStrategy | None = None, batch_length: builtins.int | None = None, auto_commit: AutoCommit | None = None, create_consumer_group_if_not_exists: builtins.bool = True, auto_join_consumer_group: builtins.bool = True, poll_interval: datetime.timedelta | None = None, polling_retry_interval: datetime.timedelta | None = None, init_retries: builtins.int | None = None, init_retry_interval: datetime.timedelta | None = None, allow_replay: builtins.bool = False) -> collections.abc.Awaitable[IggyConsumer]: + def consumer_group( + self, + name: builtins.str, + stream: builtins.str, + topic: builtins.str, + partition_id: builtins.int | None = None, + polling_strategy: PollingStrategy | None = None, + batch_length: builtins.int | None = None, + auto_commit: AutoCommit | None = None, + create_consumer_group_if_not_exists: builtins.bool = True, + auto_join_consumer_group: builtins.bool = True, + poll_interval: datetime.timedelta | None = None, + polling_retry_interval: datetime.timedelta | None = None, + init_retries: builtins.int | None = None, + init_retry_interval: datetime.timedelta | None = None, + allow_replay: builtins.bool = False, + ) -> collections.abc.Awaitable[IggyConsumer]: r""" Creates a new consumer group consumer. Returns the consumer or a PyRuntimeError on failure. @@ -465,7 +573,9 @@ class IggyConsumer: It wraps the RustIggyConsumer and provides asynchronous functionality through the contained runtime. """ - def get_last_consumed_offset(self, partition_id: builtins.int) -> builtins.int | None: + def get_last_consumed_offset( + self, partition_id: builtins.int + ) -> builtins.int | None: r""" Get the last consumed offset or `None` if no offset has been consumed yet. """ @@ -489,14 +599,18 @@ class IggyConsumer: r""" Gets the name of the topic this consumer group is configured for. """ - def store_offset(self, offset: builtins.int, partition_id: builtins.int | None) -> collections.abc.Awaitable[None]: + def store_offset( + self, offset: builtins.int, partition_id: builtins.int | None + ) -> collections.abc.Awaitable[None]: r""" Stores the provided offset for the provided partition id or if none is specified uses the current partition id for the consumer group. Returns `Ok(())` if the server responds successfully, or a `PyRuntimeError` if the operation fails. """ - def delete_offset(self, partition_id: builtins.int | None) -> collections.abc.Awaitable[None]: + def delete_offset( + self, partition_id: builtins.int | None + ) -> collections.abc.Awaitable[None]: r""" Deletes the offset for the provided partition id or if none is specified uses the current partition id for the consumer group. @@ -513,7 +627,13 @@ class IggyConsumer: only the interval part is applied; the `after` mode is ignored. Use `consume_messages()` if you need commit-after-processing semantics. """ - def consume_messages(self, callback: collections.abc.Callable[[ReceiveMessage], collections.abc.Awaitable[None]], shutdown_event: asyncio.Event | None) -> collections.abc.Awaitable[None]: + def consume_messages( + self, + callback: collections.abc.Callable[ + [ReceiveMessage], collections.abc.Awaitable[None] + ], + shutdown_event: asyncio.Event | None, + ) -> collections.abc.Awaitable[None]: r""" Consumes messages continuously using a callback function and an optional `asyncio.Event` for signaling shutdown. Returns an awaitable that completes when shutdown is signaled or a PyRuntimeError on failure. @@ -527,7 +647,9 @@ class IggyError(builtins.Exception): def name(self) -> builtins.str: ... @property def message(self) -> builtins.str: ... - def __new__(cls, code: builtins.int, name: builtins.str, message: builtins.str) -> IggyError: ... + def __new__( + cls, code: builtins.int, name: builtins.str, message: builtins.str + ) -> IggyError: ... def __str__(self) -> builtins.str: ... def __repr__(self) -> builtins.str: ... @@ -538,29 +660,29 @@ class PollingStrategy: @property def value(self) -> builtins.int: ... def __new__(cls, value: builtins.int) -> PollingStrategy.Offset: ... - + @typing.final class Timestamp(PollingStrategy): __match_args__ = ("value",) @property def value(self) -> builtins.int: ... def __new__(cls, value: builtins.int) -> PollingStrategy.Timestamp: ... - + @typing.final class First(PollingStrategy): __match_args__ = () def __new__(cls) -> PollingStrategy.First: ... - + @typing.final class Last(PollingStrategy): __match_args__ = () def __new__(cls) -> PollingStrategy.Last: ... - + @typing.final class Next(PollingStrategy): __match_args__ = () def __new__(cls) -> PollingStrategy.Next: ... - + ... @typing.final @@ -684,4 +806,3 @@ class TopicDetails: r""" Replication factor for the topic. """ - diff --git a/foreign/python/src/error.rs b/foreign/python/src/error.rs index 042f66b482..38b791ca05 100644 --- a/foreign/python/src/error.rs +++ b/foreign/python/src/error.rs @@ -20,6 +20,8 @@ use pyo3::exceptions::PyException; use pyo3::prelude::*; use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +/// A Python class representing the Rust's IggyError. +/// Allows transparent representation of Iggy-specific errors. #[gen_stub_pyclass] #[pyclass(skip_from_py_object, extends=PyException)] #[derive(Clone, Debug)] diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 6eac8fae44..1d189d6664 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -236,7 +236,7 @@ async def test_login_with_invalid_credentials_fails( result = client.login_user(username, password) await result - if expected_exception is IggyError: + if isinstance(exc_info.value, IggyError): assert exc_info.value.name == expected_iggy_error_name assert exc_info.value.code == expected_iggy_error_code assert exc_info.value.message == expected_error_message From ee7f686feaa81d1cbb942ecbb1246b09c5040d15 Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:07:46 +0200 Subject: [PATCH 4/9] feat(python): expose structured IggyError - fixes for clippy and stub update --- foreign/python/apache_iggy.pyi | 4 ++++ foreign/python/src/client.rs | 38 +++++++++++++++--------------- foreign/python/src/consumer.rs | 15 ++++++------ foreign/python/src/identifier.rs | 8 +++---- foreign/python/src/send_message.rs | 4 ++-- 5 files changed, 36 insertions(+), 33 deletions(-) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 428c09826b..05c6bad15f 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -641,6 +641,10 @@ class IggyConsumer: @typing.final class IggyError(builtins.Exception): + r""" + A Python class representing the Rust's IggyError. + Allows transparent representation of Iggy-specific errors. + """ @property def code(self) -> builtins.int: ... @property diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 8de6782983..c2725098d2 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -64,7 +64,7 @@ impl IggyClient { .with_tcp() .with_server_address(conn.unwrap_or("127.0.0.1:8090".to_string())) .build() - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(IggyClient { inner: Arc::new(client), }) @@ -81,7 +81,7 @@ impl IggyClient { connection_string: String, ) -> PyResult { let client = RustIggyClient::from_connection_string(&connection_string) - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(Self { inner: Arc::new(client), }) @@ -97,7 +97,7 @@ impl IggyClient { inner .ping() .await - .map_err(|e| PyIggyError::new_err_from_rust(e)) + .map_err(PyIggyError::new_err_from_rust) }) } @@ -115,7 +115,7 @@ impl IggyClient { inner .login_user(&username, &password) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -129,7 +129,7 @@ impl IggyClient { inner .connect() .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -144,7 +144,7 @@ impl IggyClient { inner .create_stream(&name) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -164,7 +164,7 @@ impl IggyClient { let stream = inner .get_stream(&stream_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(stream.map(StreamDetails::from)) }) } @@ -220,7 +220,7 @@ impl IggyClient { max_size, ) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -242,7 +242,7 @@ impl IggyClient { let topic = inner .get_topic(&stream_id, &topic_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(topic.map(TopicDetails::from)) }) } @@ -270,7 +270,7 @@ impl IggyClient { let topics = inner .get_topics(&stream_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(topics.into_iter().map(Topic::from).collect::>()) }) } @@ -375,7 +375,7 @@ impl IggyClient { inner .delete_topic(&stream_id, &topic_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -406,7 +406,7 @@ impl IggyClient { inner .purge_topic(&stream_id, &topic_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -440,7 +440,7 @@ impl IggyClient { inner .create_consumer_group(&stream_id, &topic_id, &name) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -476,7 +476,7 @@ impl IggyClient { let group = inner .get_consumer_group(&stream_id, &topic_id, &group_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(group.map(PyConsumerGroupDetails::from)) }) } @@ -508,7 +508,7 @@ impl IggyClient { let groups = inner .get_consumer_groups(&stream_id, &topic_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(groups .into_iter() .map(PyConsumerGroup::from) @@ -548,7 +548,7 @@ impl IggyClient { inner .send_messages(&stream, &topic, &partitioning, messages.as_mut()) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -586,7 +586,7 @@ impl IggyClient { auto_commit, ) .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; let messages = polled_messages .messages .into_iter() @@ -647,7 +647,7 @@ impl IggyClient { let mut builder = self .inner .consumer_group(name, stream, topic) - .map_err(|e| PyIggyError::new_err_from_rust(e))? + .map_err(PyIggyError::new_err_from_rust)? .without_encryptor() .partition(partition_id); @@ -705,7 +705,7 @@ impl IggyClient { consumer .init() .await - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(IggyConsumer { inner: Arc::new(Mutex::new(consumer)), }) diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 2be3f21f7b..e493c4e56e 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs @@ -39,9 +39,9 @@ use tokio::sync::Mutex; use tokio::sync::oneshot::Sender; use tokio::task::JoinHandle; +use crate::error::IggyError as PyIggyError; use crate::identifier::PyIdentifier; use crate::receive_message::ReceiveMessage; -use crate::error::IggyError as PyIggyError; /// A Python class representing the Iggy consumer. /// It wraps the RustIggyConsumer and provides asynchronous functionality @@ -111,7 +111,7 @@ impl IggyConsumer { .await .store_offset(offset, partition_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e)) + .map_err(PyIggyError::new_err_from_rust) }) } @@ -132,7 +132,7 @@ impl IggyConsumer { .await .delete_offset(partition_id) .await - .map_err(|e| PyIggyError::new_err_from_rust(e)) + .map_err(PyIggyError::new_err_from_rust) }) } @@ -168,8 +168,8 @@ impl IggyConsumer { let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); let task_locals = Python::attach(pyo3_async_runtimes::tokio::get_current_locals)?; - let handle_consume: JoinHandle>> = - get_runtime().spawn(scope(task_locals, async move { + let handle_consume: JoinHandle>> = get_runtime() + .spawn(scope(task_locals, async move { let task_locals = Python::attach(pyo3_async_runtimes::tokio::get_current_locals)?; let consumer = PyCallbackConsumer { @@ -213,7 +213,7 @@ impl IggyConsumer { consume_result .map_err(|e| PyErr::new::(e.to_string()))?? - .map_err(|e| PyIggyError::new_err_from_rust(e))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -366,8 +366,7 @@ impl ReceiveMessageIterator { inner: m.message, partition_id: m.partition_id, }) - .map_err(|e| PyIggyError::new_err_from_rust(e))? - ) + .map_err(PyIggyError::new_err_from_rust)?) } else { Err(PyStopAsyncIteration::new_err("No more messages")) } diff --git a/foreign/python/src/identifier.rs b/foreign/python/src/identifier.rs index ad7cd548c0..40e6953994 100644 --- a/foreign/python/src/identifier.rs +++ b/foreign/python/src/identifier.rs @@ -38,10 +38,10 @@ impl TryFrom for Identifier { fn try_from(py_identifier: PyIdentifier) -> Result { match py_identifier { PyIdentifier::String(s) => { - Identifier::from_str(&s).map_err(|e| PyIggyError::new_err_from_rust(e)) + Identifier::from_str(&s).map_err(PyIggyError::new_err_from_rust) } PyIdentifier::Int(i) => { - Identifier::numeric(i).map_err(|e| PyIggyError::new_err_from_rust(e)) + Identifier::numeric(i).map_err(PyIggyError::new_err_from_rust) } } } @@ -55,11 +55,11 @@ impl TryFrom<&Identifier> for PyIdentifier { IdKind::String => val .get_string_value() .map(PyIdentifier::String) - .map_err(|e| PyIggyError::new_err_from_rust(e)), + .map_err(PyIggyError::new_err_from_rust), IdKind::Numeric => val .get_u32_value() .map(PyIdentifier::Int) - .map_err(|e| PyIggyError::new_err_from_rust(e)), + .map_err(PyIggyError::new_err_from_rust), } } } diff --git a/foreign/python/src/send_message.rs b/foreign/python/src/send_message.rs index 33b11e98aa..fc24da37d2 100644 --- a/foreign/python/src/send_message.rs +++ b/foreign/python/src/send_message.rs @@ -66,14 +66,14 @@ impl SendMessage { pub fn new(py: Python, data: PyMessagePayload) -> PyResult { let inner = match data { PyMessagePayload::String(data) => { - RustIggyMessage::from_str(&data).map_err(|e| PyIggyError::new_err_from_rust(e))? + RustIggyMessage::from_str(&data).map_err(PyIggyError::new_err_from_rust)? } PyMessagePayload::Bytes(data) => { let bytes = Bytes::from(data.extract::>(py)?); RustIggyMessage::builder() .payload(bytes) .build() - .map_err(|e| PyIggyError::new_err_from_rust(e))? + .map_err(PyIggyError::new_err_from_rust)? } }; Ok(Self { inner }) From 8f2f13077b1097175c706df488652891dc24eb7d Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:45:01 +0200 Subject: [PATCH 5/9] feat(python): expose structured IggyError - minor doc update and linting updates --- foreign/python/src/client.rs | 5 +---- foreign/python/src/error.rs | 1 + foreign/python/src/identifier.rs | 4 +--- 3 files changed, 3 insertions(+), 7 deletions(-) diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index c2725098d2..27db364e21 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -94,10 +94,7 @@ impl IggyClient { fn ping<'a>(&self, py: Python<'a>) -> PyResult> { let inner = self.inner.clone(); future_into_py(py, async move { - inner - .ping() - .await - .map_err(PyIggyError::new_err_from_rust) + inner.ping().await.map_err(PyIggyError::new_err_from_rust) }) } diff --git a/foreign/python/src/error.rs b/foreign/python/src/error.rs index 38b791ca05..d7066bea55 100644 --- a/foreign/python/src/error.rs +++ b/foreign/python/src/error.rs @@ -22,6 +22,7 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; /// A Python class representing the Rust's IggyError. /// Allows transparent representation of Iggy-specific errors. +/// You can find all available error codes and names in IggyError enum, which lives in core/common/src/error/iggy_error.rs. #[gen_stub_pyclass] #[pyclass(skip_from_py_object, extends=PyException)] #[derive(Clone, Debug)] diff --git a/foreign/python/src/identifier.rs b/foreign/python/src/identifier.rs index 40e6953994..4b1d6d9192 100644 --- a/foreign/python/src/identifier.rs +++ b/foreign/python/src/identifier.rs @@ -40,9 +40,7 @@ impl TryFrom for Identifier { PyIdentifier::String(s) => { Identifier::from_str(&s).map_err(PyIggyError::new_err_from_rust) } - PyIdentifier::Int(i) => { - Identifier::numeric(i).map_err(PyIggyError::new_err_from_rust) - } + PyIdentifier::Int(i) => Identifier::numeric(i).map_err(PyIggyError::new_err_from_rust), } } } From 3803387a5c0aebf1c43468ea7a0ebeb47fb9e37c Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Sat, 18 Jul 2026 01:46:06 +0200 Subject: [PATCH 6/9] feat(python): expose structured IggyError - docs and examples updates --- examples/python/basic/consumer.py | 8 +++++++- examples/python/basic/producer.py | 15 ++++++++++++++- examples/python/getting-started/consumer.py | 12 +++++++++++- examples/python/getting-started/producer.py | 15 ++++++++++++++- foreign/python/src/error.rs | 18 +++++++++++++++--- 5 files changed, 61 insertions(+), 7 deletions(-) diff --git a/examples/python/basic/consumer.py b/examples/python/basic/consumer.py index e20cf3175b..cf35faefb7 100644 --- a/examples/python/basic/consumer.py +++ b/examples/python/basic/consumer.py @@ -19,7 +19,7 @@ import asyncio from typing import NamedTuple -from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage +from apache_iggy import IggyClient, IggyError, PollingStrategy, ReceiveMessage from loguru import logger STREAM_NAME = "sample-stream" @@ -88,6 +88,12 @@ async def consume_messages(client: IggyClient): handle_message(message) n_consumed_batches += 1 await asyncio.sleep(interval) + except IggyError as error: + logger.error( + "Iggy error while consuming messages: " + f"[{error.code}] {error.name}: {error.message}" + ) + break except Exception as error: logger.exception(f"Exception occurred while consuming messages: {error}") break diff --git a/examples/python/basic/producer.py b/examples/python/basic/producer.py index b04ca8fc86..5566f4d08d 100644 --- a/examples/python/basic/producer.py +++ b/examples/python/basic/producer.py @@ -19,7 +19,7 @@ import asyncio from typing import NamedTuple -from apache_iggy import IggyClient, StreamDetails, TopicDetails +from apache_iggy import IggyClient, IggyError, StreamDetails, TopicDetails from apache_iggy import SendMessage as Message from loguru import logger @@ -69,6 +69,10 @@ async def init_system(client: IggyClient): else: logger.warning(f"Stream {stream.name} already exists with ID {stream.id}") + except IggyError as error: + logger.error( + f"Error creating stream: [{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.error(f"Error creating stream: {error}") logger.exception(error) @@ -86,6 +90,10 @@ async def init_system(client: IggyClient): logger.info("Topic was created successfully.") else: logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + except IggyError as error: + logger.error( + f"Error creating topic: [{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.error(f"Error creating topic {error}") logger.exception(error) @@ -124,6 +132,11 @@ async def produce_messages(client: IggyClient): f"Successfully sent batch of {messages_per_batch} messages. " f"Batch ID: {current_id // messages_per_batch}" ) + except IggyError as error: + logger.error( + "Iggy error while sending messages: " + f"[{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.error(f"Exception type: {type(error).__name__}, message: {error}") logger.exception(error) diff --git a/examples/python/getting-started/consumer.py b/examples/python/getting-started/consumer.py index bb10e91382..b60f01386d 100755 --- a/examples/python/getting-started/consumer.py +++ b/examples/python/getting-started/consumer.py @@ -20,7 +20,7 @@ import typing import urllib.parse -from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage +from apache_iggy import IggyClient, IggyError, PollingStrategy, ReceiveMessage from loguru import logger STREAM_NAME = "sample-stream" @@ -122,6 +122,10 @@ async def main(): await client.connect() logger.info("Connected.") await consume_messages(client) + except IggyError as error: + logger.error( + f"Iggy error in main: [{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.exception(f"Exception occurred in main function: {error}") @@ -157,6 +161,12 @@ async def consume_messages(client: IggyClient): handle_message(message) n_consumed_batches += 1 await asyncio.sleep(interval) + except IggyError as error: + logger.error( + "Iggy error while consuming messages: " + f"[{error.code}] {error.name}: {error.message}" + ) + break except Exception as error: logger.exception(f"Exception occurred while consuming messages: {error}") break diff --git a/examples/python/getting-started/producer.py b/examples/python/getting-started/producer.py index c05c0317cd..6cb95f2231 100755 --- a/examples/python/getting-started/producer.py +++ b/examples/python/getting-started/producer.py @@ -20,7 +20,7 @@ import typing import urllib.parse -from apache_iggy import IggyClient, StreamDetails, TopicDetails +from apache_iggy import IggyClient, IggyError, StreamDetails, TopicDetails from apache_iggy import SendMessage as Message from loguru import logger @@ -135,6 +135,10 @@ async def init_system(client: IggyClient): else: logger.warning(f"Stream {stream.name} already exists with ID {stream.id}") + except IggyError as error: + logger.error( + f"Error creating stream: [{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.error(f"Error creating stream: {error}") logger.exception(error) @@ -152,6 +156,10 @@ async def init_system(client: IggyClient): logger.info("Topic was created successfully.") else: logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + except IggyError as error: + logger.error( + f"Error creating topic: [{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.error(f"Error creating topic {error}") logger.exception(error) @@ -190,6 +198,11 @@ async def produce_messages(client: IggyClient): f"Successfully sent batch of {messages_per_batch} messages. " f"Batch ID: {current_id // messages_per_batch}" ) + except IggyError as error: + logger.error( + "Iggy error while sending messages: " + f"[{error.code}] {error.name}: {error.message}" + ) except Exception as error: logger.error(f"Exception type: {type(error).__name__}, message: {error}") logger.exception(error) diff --git a/foreign/python/src/error.rs b/foreign/python/src/error.rs index d7066bea55..c0208913c4 100644 --- a/foreign/python/src/error.rs +++ b/foreign/python/src/error.rs @@ -15,22 +15,32 @@ // specific language governing permissions and limitations // under the License. +//! Python-facing error type bridging [`iggy::prelude::IggyError`] to a +//! `PyException` subclass so callers can `except IggyError as e` and +//! inspect `e.code`, `e.name`, and `e.message` directly. +//! +//! Error codes and names are defined in +//! `core/common/src/error/iggy_error.rs`. + use iggy::prelude::IggyError as RustIggyError; use pyo3::exceptions::PyException; use pyo3::prelude::*; use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; -/// A Python class representing the Rust's IggyError. -/// Allows transparent representation of Iggy-specific errors. -/// You can find all available error codes and names in IggyError enum, which lives in core/common/src/error/iggy_error.rs. +/// Iggy-specific exception exposed to Python. #[gen_stub_pyclass] #[pyclass(skip_from_py_object, extends=PyException)] #[derive(Clone, Debug)] pub struct IggyError { + /// Numeric error code from [`iggy::prelude::IggyError::as_code`]. #[pyo3(get)] code: u32, + /// Snake-case error variant name from [`iggy::prelude::IggyError::as_string`] + /// (e.g. `"topic_name_already_exists"`). #[pyo3(get)] name: String, + /// Human-readable description from the variant's `#[error(...)]` template + /// (e.g. `"Topic with name: foo for stream with ID: 1 already exists."`). #[pyo3(get)] message: String, } @@ -60,6 +70,8 @@ impl IggyError { } impl IggyError { + /// Convert a Rust [`RustIggyError`] into a [`PyErr`] carrying this type. + /// Use this at every SDK boundary instead of mapping to a generic `PyRuntimeError`. pub fn new_err_from_rust(e: RustIggyError) -> PyErr { PyErr::new::((e.as_code(), e.as_string().to_string(), e.to_string())) } From ed1e77b26bbd3bc4101e5ef7c5b7f905e6e562b0 Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Sat, 18 Jul 2026 02:38:00 +0200 Subject: [PATCH 7/9] feat(python): expose structured IggyError - test updates --- foreign/python/tests/test_connectivity.py | 26 +++---------------- foreign/python/tests/test_consumer_group.py | 14 ---------- .../python/tests/test_message_operations.py | 3 --- foreign/python/tests/test_stream.py | 6 ----- foreign/python/tests/test_topic.py | 19 -------------- 5 files changed, 3 insertions(+), 65 deletions(-) diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 1d189d6664..6301086483 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -56,89 +56,75 @@ async def test_valid_connection_string(self, connection_string: str): "invalid_value", "expected_iggy_error_name", "expected_iggy_error_code", - "expected_iggy_error_message", ), [ - ("", "invalid_connection_string", 8000, "Invalid connection string"), + ("", "invalid_connection_string", 8000), ( "bad address", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "http://{host}:{port}", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "tcp://iggy:iggy@{host}:{port}", "invalid_connection_string", 8000, - "Invalid connection string", ), - ("{host}:", "invalid_connection_string", 8000, "Invalid connection string"), - (":{port}", "invalid_connection_string", 8000, "Invalid connection string"), + ("{host}:", "invalid_connection_string", 8000), + (":{port}", "invalid_connection_string", 8000), ( "{host}:not-a-port", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "{host}:70000", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:not-a-port", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:-1", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:70000", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://iggy:bad:format@{host}:{port}", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value", "invalid_connection_string", 8000, - "Invalid connection string", ), ( "iggy+quic://iggy:iggy@127.0.0.1:8080", "cannot_create_endpoint", 305, - "Cannot create endpoint", ), ], ) @@ -147,7 +133,6 @@ def test_invalid_connection_string( invalid_value: str, expected_iggy_error_name: str, expected_iggy_error_code: int, - expected_iggy_error_message: str, ): """Test malformed server addresses and connection strings are rejected.""" host, port = get_server_config() @@ -158,9 +143,6 @@ def test_invalid_connection_string( assert excinfo_from_connection_string.value.name == expected_iggy_error_name assert excinfo_from_connection_string.value.code == expected_iggy_error_code - assert ( - excinfo_from_connection_string.value.message == expected_iggy_error_message - ) @pytest.mark.asyncio async def test_repeated_connect_does_not_error(self): @@ -239,7 +221,6 @@ async def test_login_with_invalid_credentials_fails( if isinstance(exc_info.value, IggyError): assert exc_info.value.name == expected_iggy_error_name assert exc_info.value.code == expected_iggy_error_code - assert exc_info.value.message == expected_error_message else: assert exc_info.value.args[0] == expected_error_message @@ -259,7 +240,6 @@ async def test_relogin_with_invalid_credentials_fails(self): assert exc_info.value.name == "invalid_credentials" assert exc_info.value.code == 42 - assert exc_info.value.message == "Invalid credentials" @pytest.mark.asyncio async def test_relogin_with_same_valid_credentials_does_not_error(self): diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py index e61745a571..f59c0f2834 100644 --- a/foreign/python/tests/test_consumer_group.py +++ b/foreign/python/tests/test_consumer_group.py @@ -172,7 +172,6 @@ async def test_duplicate_consumer_group_creation_fails( assert exc_info.value.name == "consumer_group_name_already_exists" assert exc_info.value.code == 5004 - assert exc_info.value.message.startswith("Consumer group with name:") @pytest.mark.asyncio async def test_create_consumer_group_requires_connection_and_auth( @@ -192,7 +191,6 @@ async def test_create_consumer_group_requires_connection_and_auth( assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -204,7 +202,6 @@ async def test_create_consumer_group_requires_connection_and_auth( assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestGetConsumerGroup: @@ -340,7 +337,6 @@ async def test_get_consumer_group_requires_connection_and_auth(self, unique_name assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -352,7 +348,6 @@ async def test_get_consumer_group_requires_connection_and_auth(self, unique_name assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestGetConsumerGroups: @@ -484,7 +479,6 @@ async def test_get_consumer_groups_requires_connection_and_auth(self, unique_nam assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -492,7 +486,6 @@ async def test_get_consumer_groups_requires_connection_and_auth(self, unique_nam assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestDeleteConsumerGroup: @@ -1070,10 +1063,6 @@ async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name assert exc_info.value.name == "consumer_offset_not_found" assert exc_info.value.code == 3021 - assert ( - exc_info.value.message - == "Consumer offset for consumer with ID: 0 was not found." - ) assert consumer.get_last_stored_offset(partition_id) is None await iggy_client.send_messages( @@ -2031,7 +2020,6 @@ async def test_consumer_group_auto_create_disabled_fails_for_missing_group( assert exc_info.value.name == "consumer_group_name_not_found" assert exc_info.value.code == 5003 - assert exc_info.value.message.startswith("Consumer group with name:") @pytest.mark.asyncio @pytest.mark.parametrize( @@ -2094,7 +2082,6 @@ async def test_consumer_group_before_connect_fails(self, unique_name): assert exc_info.value.name == "disconnected" assert exc_info.value.code == 8 - assert exc_info.value.message == "Disconnected" @pytest.mark.asyncio async def test_consumer_group_before_login_fails(self, unique_name): @@ -2120,7 +2107,6 @@ async def test_consumer_group_before_login_fails(self, unique_name): assert exc_info.value.name == "unauthenticated" assert exc_info.value.code == 40 - assert exc_info.value.message == "Unauthenticated" @pytest.mark.asyncio async def test_shutdown(self, iggy_client: IggyClient, unique_name): diff --git a/foreign/python/tests/test_message_operations.py b/foreign/python/tests/test_message_operations.py index cf02bf755e..f9b36bfa20 100644 --- a/foreign/python/tests/test_message_operations.py +++ b/foreign/python/tests/test_message_operations.py @@ -151,7 +151,6 @@ async def test_empty_payload_is_rejected(self, payload): assert exc_info.value.name == "invalid_message_payload_length" assert exc_info.value.code == 4025 - assert exc_info.value.message == "Invalid message payload length" @pytest.mark.asyncio @pytest.mark.parametrize( @@ -377,7 +376,6 @@ async def test_poll_messages_with_count_zero_is_rejected( assert exc_info.value.name == "invalid_messages_count" assert exc_info.value.code == 4009 - assert exc_info.value.message == "Invalid messages count" @pytest.mark.asyncio async def test_poll_messages_with_invalid_partition_id_raises( @@ -412,7 +410,6 @@ async def test_poll_messages_with_invalid_partition_id_raises( assert exc_info.value.name == "partition_not_found" assert exc_info.value.code == 3007 - assert exc_info.value.message.startswith("Partition with ID:") @pytest.mark.asyncio async def test_polling_strategy_last_with_count_one_returns_last_message( diff --git a/foreign/python/tests/test_stream.py b/foreign/python/tests/test_stream.py index 64595fc2c9..a5b0db8fcc 100644 --- a/foreign/python/tests/test_stream.py +++ b/foreign/python/tests/test_stream.py @@ -96,7 +96,6 @@ async def test_create_stream_invalid_names( assert exc_info.value.name == "invalid_format" assert exc_info.value.code == 4 - assert exc_info.value.message == "Invalid format" @pytest.mark.asyncio async def test_get_stream_by_name_and_id( @@ -155,7 +154,6 @@ async def test_duplicate_stream_creation( assert exc_info.value.name == "stream_name_already_exists" assert exc_info.value.code == 1012 - assert exc_info.value.message.startswith("Stream with name:") @pytest.mark.asyncio async def test_get_nonexistent_stream(self, iggy_client: IggyClient, unique_name): @@ -179,7 +177,6 @@ async def test_get_stream_before_connect_fails(self, unique_name): assert exc_info.value.name == "disconnected" assert exc_info.value.code == 8 - assert exc_info.value.message == "Disconnected" @pytest.mark.asyncio async def test_get_stream_before_login_fails(self, unique_name): @@ -195,7 +192,6 @@ async def test_get_stream_before_login_fails(self, unique_name): assert exc_info.value.name == "unauthenticated" assert exc_info.value.code == 40 - assert exc_info.value.message == "Unauthenticated" @pytest.mark.asyncio async def test_create_stream_before_connect_fails(self, unique_name): @@ -208,7 +204,6 @@ async def test_create_stream_before_connect_fails(self, unique_name): assert exc_info.value.name == "disconnected" assert exc_info.value.code == 8 - assert exc_info.value.message == "Disconnected" @pytest.mark.asyncio async def test_create_stream_before_login_fails(self, unique_name): @@ -224,4 +219,3 @@ async def test_create_stream_before_login_fails(self, unique_name): assert exc_info.value.name == "unauthenticated" assert exc_info.value.code == 40 - assert exc_info.value.message == "Unauthenticated" diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py index c2823d7981..c2a32e82a5 100644 --- a/foreign/python/tests/test_topic.py +++ b/foreign/python/tests/test_topic.py @@ -111,7 +111,6 @@ async def test_create_topic_invalid_names( assert exc_info.value.name == "invalid_format" assert exc_info.value.code == 4 - assert exc_info.value.message == "Invalid format" @pytest.mark.asyncio async def test_create_and_get_topic_with_numeric_stream_id( @@ -157,7 +156,6 @@ async def test_duplicate_topic_creation(self, iggy_client: IggyClient, unique_na assert exc_info.value.name == "topic_name_already_exists" assert exc_info.value.code == 2013 - assert exc_info.value.message.startswith("Topic with name:") @pytest.mark.asyncio async def test_topic_names_can_repeat_across_different_streams( @@ -341,7 +339,6 @@ async def test_create_topic_invalid_max_topic_size( if expected_exception is IggyError: assert exc_info.value.name == "invalid_topic_size" assert exc_info.value.code == 1019 - assert exc_info.value.message.startswith("Max topic size cannot") @pytest.mark.asyncio @pytest.mark.parametrize( @@ -421,7 +418,6 @@ async def test_create_topic_invalid_partitions_count( assert exc_info.value.name == "too_many_partitions" assert exc_info.value.code == 2015 - assert exc_info.value.message == "Too many partitions" @pytest.mark.asyncio @pytest.mark.parametrize("partitions_count", [0, 1, 1000]) @@ -509,7 +505,6 @@ async def test_create_topic_in_nonexistent_stream( assert exc_info.value.name == "stream_id_not_found" assert exc_info.value.code == 1009 - assert exc_info.value.message.startswith("Stream with ID:") @pytest.mark.asyncio async def test_create_topic_requires_connection_and_auth(self, unique_name): @@ -525,7 +520,6 @@ async def test_create_topic_requires_connection_and_auth(self, unique_name): assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -535,7 +529,6 @@ async def test_create_topic_requires_connection_and_auth(self, unique_name): assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestGetTopic: @@ -602,7 +595,6 @@ async def test_get_topic_requires_connection_and_auth(self, unique_name): assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -610,7 +602,6 @@ async def test_get_topic_requires_connection_and_auth(self, unique_name): assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestGetTopics: @@ -717,7 +708,6 @@ async def test_get_topics_requires_connection_and_auth(self, unique_name): assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -725,7 +715,6 @@ async def test_get_topics_requires_connection_and_auth(self, unique_name): assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestUpdateTopic: @@ -1196,7 +1185,6 @@ async def test_delete_nonexistent_topic_fails( assert exc_info.value.name == "topic_id_not_found" assert exc_info.value.code == 2010 - assert exc_info.value.message.startswith("Topic with ID:") @pytest.mark.asyncio async def test_delete_topic_twice_fails_second_time( @@ -1217,7 +1205,6 @@ async def test_delete_topic_twice_fails_second_time( assert exc_info.value.name == "topic_id_not_found" assert exc_info.value.code == 2010 - assert exc_info.value.message.startswith("Topic with ID:") @pytest.mark.asyncio async def test_delete_topic_requires_connection_and_auth(self, unique_name): @@ -1231,7 +1218,6 @@ async def test_delete_topic_requires_connection_and_auth(self, unique_name): assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -1239,7 +1225,6 @@ async def test_delete_topic_requires_connection_and_auth(self, unique_name): assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" class TestPurgeTopic: @@ -1333,7 +1318,6 @@ async def test_purge_nonexistent_topic_fails( assert exc_info.value.name == "topic_id_not_found" assert exc_info.value.code == 2010 - assert exc_info.value.message.startswith("Topic with ID:") @pytest.mark.asyncio async def test_purge_topic_in_nonexistent_stream_fails( @@ -1345,7 +1329,6 @@ async def test_purge_topic_in_nonexistent_stream_fails( assert exc_info.value.name == "stream_id_not_found" assert exc_info.value.code == 1009 - assert exc_info.value.message.startswith("Stream with ID:") @pytest.mark.asyncio async def test_purge_topic_requires_connection_and_auth(self, unique_name): @@ -1359,7 +1342,6 @@ async def test_purge_topic_requires_connection_and_auth(self, unique_name): assert exc_info_disconnected.value.name == "disconnected" assert exc_info_disconnected.value.code == 8 - assert exc_info_disconnected.value.message == "Disconnected" await client.connect() with pytest.raises(IggyError) as exc_info_unauthenticated: @@ -1367,4 +1349,3 @@ async def test_purge_topic_requires_connection_and_auth(self, unique_name): assert exc_info_unauthenticated.value.name == "unauthenticated" assert exc_info_unauthenticated.value.code == 40 - assert exc_info_unauthenticated.value.message == "Unauthenticated" From 6eaaaec00006f240efbb62f3f5479564cedaad90 Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Sat, 18 Jul 2026 12:09:59 +0200 Subject: [PATCH 8/9] feat(python): expose structured IggyError - fixed missed exceptions, updated code after merging --- foreign/python/src/client.rs | 8 ++--- foreign/python/tests/test_consumer_group.py | 33 ++++++++++++++++----- foreign/python/tests/test_topic.py | 25 ++++++++++++---- 3 files changed, 49 insertions(+), 17 deletions(-) diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 6da5e545db..50fc2e244f 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -341,7 +341,7 @@ impl IggyClient { max_size, ) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -543,7 +543,7 @@ impl IggyClient { inner .delete_consumer_group(&stream_id, &topic_id, &group_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -581,7 +581,7 @@ impl IggyClient { inner .join_consumer_group(&stream_id, &topic_id, &group_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } @@ -621,7 +621,7 @@ impl IggyClient { inner .leave_consumer_group(&stream_id, &topic_id, &group_id) .await - .map_err(|e| PyErr::new::(e.to_string()))?; + .map_err(PyIggyError::new_err_from_rust)?; Ok(()) }) } diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py index f59c0f2834..8915525722 100644 --- a/foreign/python/tests/test_consumer_group.py +++ b/foreign/python/tests/test_consumer_group.py @@ -573,13 +573,16 @@ async def test_delete_nonexistent_consumer_group_fails( partitions_count=1, ) - with pytest.raises(RuntimeError, match="Consumer group with ID:"): + with pytest.raises(IggyError) as exc_info: await iggy_client.delete_consumer_group( stream_name, topic_name, unique_name(), ) + assert exc_info.value.code == 5000 + assert exc_info.value.name == "consumer_group_id_not_found" + @pytest.mark.asyncio async def test_delete_consumer_group_removes_live_member( self, iggy_client: IggyClient, unique_name @@ -696,13 +699,16 @@ async def test_join_nonexistent_consumer_group_fails( partitions_count=1, ) - with pytest.raises(RuntimeError, match="Consumer group with ID:"): + with pytest.raises(IggyError) as exc_info: await iggy_client.join_consumer_group( stream_name, topic_name, unique_name(), ) + assert exc_info.value.code == 5000 + assert exc_info.value.name == "consumer_group_id_not_found" + class TestLeaveConsumerGroup: """Test leaving consumer groups via leave_consumer_group.""" @@ -786,11 +792,13 @@ async def test_leave_consumer_group_twice_fails_second_time( await iggy_client.leave_consumer_group(stream_name, topic_name, group_name) with pytest.raises( - RuntimeError, - match="Consumer group member with client ID:", - ): + IggyError, + ) as exc_info: await iggy_client.leave_consumer_group(stream_name, topic_name, group_name) + assert exc_info.value.code == 5006 + assert exc_info.value.name == "consumer_group_member_not_found" + @pytest.mark.asyncio async def test_leave_nonexistent_consumer_group_fails( self, iggy_client: IggyClient, unique_name @@ -806,13 +814,16 @@ async def test_leave_nonexistent_consumer_group_fails( partitions_count=1, ) - with pytest.raises(RuntimeError, match="Consumer group with ID:"): + with pytest.raises(IggyError) as exc_info: await iggy_client.leave_consumer_group( stream_name, topic_name, unique_name(), ) + assert exc_info.value.code == 5000 + assert exc_info.value.name == "consumer_group_id_not_found" + @pytest.mark.parametrize( "method_name", @@ -834,13 +845,19 @@ async def test_consumer_group_lifecycle_requires_connection_and_auth( method = getattr(client, method_name) identifiers = (unique_name(), unique_name(), unique_name()) - with pytest.raises(RuntimeError, match="Disconnected"): + with pytest.raises(IggyError) as exc_info_before_conn: await method(*identifiers) + assert exc_info_before_conn.value.code == 8 + assert exc_info_before_conn.value.name == "disconnected" + await client.connect() - with pytest.raises(RuntimeError, match="Unauthenticated"): + with pytest.raises(IggyError) as exc_info_after_conn: await method(*identifiers) + assert exc_info_after_conn.value.code == 40 + assert exc_info_after_conn.value.name == "unauthenticated" + class TestConsumerGroup: """Test consumer group consumers.""" diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py index c2a32e82a5..8d22312580 100644 --- a/foreign/python/tests/test_topic.py +++ b/foreign/python/tests/test_topic.py @@ -1048,21 +1048,27 @@ async def test_update_nonexistent_topic_fails( await iggy_client.create_stream(stream_name) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.update_topic( stream_id=stream_name, topic_id=unique_name(), name=unique_name() ) + assert exc_info.value.code == 2010 + assert exc_info.value.name == "topic_id_not_found" + @pytest.mark.asyncio async def test_update_topic_in_nonexistent_stream_fails( self, iggy_client: IggyClient, unique_name ): """Test update_topic raises when the stream does not exist.""" - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.update_topic( stream_id=unique_name(), topic_id=unique_name(), name=unique_name() ) + assert exc_info.value.code == 1009 + assert exc_info.value.name == "stream_id_not_found" + @pytest.mark.asyncio async def test_update_topic_to_existing_name_fails( self, iggy_client: IggyClient, unique_name @@ -1080,11 +1086,14 @@ async def test_update_topic_to_existing_name_fails( stream=stream_name, name=second_topic, partitions_count=1 ) - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info: await iggy_client.update_topic( stream_id=stream_name, topic_id=second_topic, name=first_topic ) + assert exc_info.value.code == 2013 + assert exc_info.value.name == "topic_name_already_exists" + @pytest.mark.asyncio async def test_update_topic_requires_connection_and_auth(self, unique_name): """Test update_topic fails both before connecting and before logging in.""" @@ -1092,17 +1101,23 @@ async def test_update_topic_requires_connection_and_auth(self, unique_name): wait_for_server(host, port) client = IggyClient(f"{host}:{port}") - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_before_conn: await client.update_topic( stream_id=unique_name(), topic_id=unique_name(), name=unique_name() ) + assert exc_info_before_conn.value.code == 8 + assert exc_info_before_conn.value.name == "disconnected" + await client.connect() - with pytest.raises(RuntimeError): + with pytest.raises(IggyError) as exc_info_after_conn: await client.update_topic( stream_id=unique_name(), topic_id=unique_name(), name=unique_name() ) + assert exc_info_after_conn.value.code == 40 + assert exc_info_after_conn.value.name == "unauthenticated" + class TestDeleteTopic: """Test deleting topics via delete_topic.""" From 8d2402f8f70a1828886cd33fa8da1230cc77a0ef Mon Sep 17 00:00:00 2001 From: mmmmxa <36201581+mmmmxa@users.noreply.github.com> Date: Sat, 18 Jul 2026 12:19:47 +0200 Subject: [PATCH 9/9] feat(python): expose structured IggyError - small test fix --- foreign/python/tests/test_connectivity.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 6301086483..c0c1ab50bb 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -190,8 +190,8 @@ async def test_login_before_connect_does_not_error(self): 42, "Invalid credentials", ), - ("", "iggy", IggyError, "invalid_username", 43, "Invalid username"), - ("iggy", "", IggyError, "invalid_password", 44, "Invalid password"), + ("", "iggy", IggyError, "invalid_username", 43, None), + ("iggy", "", IggyError, "invalid_password", 44, None), (None, "iggy", TypeError, None, None, "'None' is not an instance of 'str'"), ("iggy", None, TypeError, None, None, "'None' is not an instance of 'str'"), ],