This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 1ac819e99 feat(python): expose TCP client configuration (#3776)
1ac819e99 is described below
commit 1ac819e996ac1764fd966c8a6de4dc057a4bca96
Author: Ethan Lin <[email protected]>
AuthorDate: Mon Aug 10 01:53:15 2026 -0700
feat(python): expose TCP client configuration (#3776)
---
core/sdk/src/prelude.rs | 24 +-
examples/python/getting-started/consumer.py | 52 ++--
examples/python/getting-started/producer.py | 50 +--
foreign/python/Cargo.toml | 1 +
foreign/python/README.md | 36 +++
foreign/python/apache_iggy.pyi | 263 ++++++++++++----
foreign/python/src/client.rs | 135 ++++----
foreign/python/src/config.rs | 423 +++++++++++++++++++++++++
foreign/python/src/consumer.rs | 51 +---
foreign/python/src/duration.rs | 65 ++++
foreign/python/src/lib.rs | 8 +-
foreign/python/src/receive_message.rs | 2 +-
foreign/python/src/send_message.rs | 2 -
foreign/python/src/topic.rs | 33 +-
foreign/python/tests/conftest.py | 5 +
foreign/python/tests/test_client_config.py | 459 ++++++++++++++++++++++++++++
16 files changed, 1386 insertions(+), 223 deletions(-)
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index ee0dc06d6..51b34e34a 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -51,18 +51,18 @@ pub use iggy_common::{
Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, CacheMetrics,
CacheMetricsKey, ClientError,
ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole,
ClusterNodeStatus,
CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails,
ConsumerGroupMember,
- ConsumerKind, EncryptorKind, GlobalPermissions, HeaderField, HeaderKey,
HeaderKind,
- HeaderValue, HttpClientConfig, HttpClientConfigBuilder, HttpMethod,
IdKind, Identifier,
- IdentityInfo, IggyByteSize, IggyDuration, IggyError, IggyExpiry,
IggyIndexView, IggyMessage,
- IggyMessageHeader, IggyMessageHeaderView, IggyMessageView,
IggyMessageViewIterator,
- IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning,
Permissions,
- PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind,
PollingStrategy,
- QuicClientConfig, QuicClientConfigBuilder, QuicClientReconnectionConfig,
SendMessages,
- SendMessagesConfirmationResponse, SendMessagesResponse, Sizeable,
SnapshotCompression, Stats,
- Stream, StreamDetails, StreamPermissions, SystemSnapshotType,
TcpClientConfig,
- TcpClientConfigBuilder, TcpClientReconnectionConfig, Topic, TopicDetails,
TopicPermissions,
- TransportEndpoints, TransportProtocol, UserId, UserInfo, UserInfoDetails,
UserStatus,
- Validatable, WebSocketClientConfig, WebSocketClientConfigBuilder,
+ ConsumerKind, Credentials, EncryptorKind, GlobalPermissions, HeaderField,
HeaderKey,
+ HeaderKind, HeaderValue, HttpClientConfig, HttpClientConfigBuilder,
HttpMethod, IdKind,
+ Identifier, IdentityInfo, IggyByteSize, IggyDuration, IggyError,
IggyExpiry, IggyIndexView,
+ IggyMessage, IggyMessageHeader, IggyMessageHeaderView, IggyMessageView,
+ IggyMessageViewIterator, IggyTimestamp, MaxTopicSize, Partition,
Partitioner, Partitioning,
+ Permissions, PersonalAccessTokenExpiry, PollMessages, PolledMessages,
PollingKind,
+ PollingStrategy, QuicClientConfig, QuicClientConfigBuilder,
QuicClientReconnectionConfig,
+ SendMessages, SendMessagesConfirmationResponse, SendMessagesResponse,
Sizeable,
+ SnapshotCompression, Stats, Stream, StreamDetails, StreamPermissions,
SystemSnapshotType,
+ TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig,
Topic, TopicDetails,
+ TopicPermissions, TransportEndpoints, TransportProtocol, UserId, UserInfo,
UserInfoDetails,
+ UserStatus, Validatable, WebSocketClientConfig,
WebSocketClientConfigBuilder,
WebSocketClientReconnectionConfig, defaults, locking,
};
pub use iggy_common::{
diff --git a/examples/python/getting-started/consumer.py
b/examples/python/getting-started/consumer.py
index bb10e9138..db0a6edb1 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -19,8 +19,16 @@ import argparse
import asyncio
import typing
import urllib.parse
-
-from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage
+from datetime import timedelta
+
+from apache_iggy import (
+ AutoLogin,
+ IggyClient,
+ PollingStrategy,
+ ReceiveMessage,
+ TcpConfig,
+ TcpReconnectionConfig,
+)
from loguru import logger
STREAM_NAME = "sample-stream"
@@ -91,34 +99,34 @@ def parse_args() -> ArgNamespace:
return ArgNamespace(**vars(args))
-def build_connection_string(args) -> str:
- """Build a connection string with TLS support."""
-
- conn_str =
f"iggy://{args.username}:{args.password}@{args.tcp_server_address}"
-
- if args.tls:
- # Extract domain from server address (host:port -> host)
- host = args.tcp_server_address.split(":")[0]
- query_params = ["tls=true", f"tls_domain={host}"]
-
- # Add CA file if provided
- if args.tls_ca_file:
- query_params.append(f"tls_ca_file={args.tls_ca_file}")
- conn_str += "?" + "&".join(query_params)
+def build_config(args: ArgNamespace) -> TcpConfig:
+ """Build a TCP client configuration with auto-login and reconnection."""
- return conn_str
+ return TcpConfig(
+ server_address=args.tcp_server_address,
+ auto_login=AutoLogin.username_password(args.username, args.password),
+ reconnection=TcpReconnectionConfig(
+ enabled=True,
+ interval=timedelta(seconds=1),
+ ),
+ tls_enabled=args.tls,
+ tls_ca_file=args.tls_ca_file or None,
+ )
async def main():
args: ArgNamespace = parse_args()
+ try:
+ config = build_config(args)
+ except ValueError as error:
+ logger.error(f"Invalid client configuration: {error}")
+ return
+ logger.info(f"Connecting to {args.tcp_server_address} (TLS: {args.tls})")
- # Build connection string with TLS support
- connection_string = build_connection_string(args)
- logger.info(f"Connection string: {connection_string}")
-
- client = IggyClient.from_connection_string(connection_string)
+ client = IggyClient(config)
try:
logger.info("Connecting to IggyClient...")
+ # No login_user() call: auto_login replays the credentials on every
connect.
await client.connect()
logger.info("Connected.")
await consume_messages(client)
diff --git a/examples/python/getting-started/producer.py
b/examples/python/getting-started/producer.py
index 642399edb..23f964fa8 100755
--- a/examples/python/getting-started/producer.py
+++ b/examples/python/getting-started/producer.py
@@ -19,8 +19,16 @@ import argparse
import asyncio
import typing
import urllib.parse
-
-from apache_iggy import IggyClient, StreamDetails, TopicDetails
+from datetime import timedelta
+
+from apache_iggy import (
+ AutoLogin,
+ IggyClient,
+ StreamDetails,
+ TcpConfig,
+ TcpReconnectionConfig,
+ TopicDetails,
+)
from apache_iggy import SendMessage as Message
from loguru import logger
@@ -92,33 +100,33 @@ def parse_args() -> ArgNamespace:
return ArgNamespace(**vars(args))
-def build_connection_string(args) -> str:
- """Build a connection string with TLS support."""
-
- conn_str =
f"iggy://{args.username}:{args.password}@{args.tcp_server_address}"
-
- if args.tls:
- # Extract domain from server address (host:port -> host)
- host = args.tcp_server_address.split(":")[0]
- query_params = ["tls=true", f"tls_domain={host}"]
+def build_config(args: ArgNamespace) -> TcpConfig:
+ """Build a TCP client configuration with auto-login and reconnection."""
- # Add CA file if provided
- if args.tls_ca_file:
- query_params.append(f"tls_ca_file={args.tls_ca_file}")
- conn_str += "?" + "&".join(query_params)
-
- return conn_str
+ return TcpConfig(
+ server_address=args.tcp_server_address,
+ auto_login=AutoLogin.username_password(args.username, args.password),
+ reconnection=TcpReconnectionConfig(
+ enabled=True,
+ interval=timedelta(seconds=1),
+ ),
+ tls_enabled=args.tls,
+ tls_ca_file=args.tls_ca_file or None,
+ )
async def main():
args: ArgNamespace = parse_args()
- # Build connection string with TLS support
- connection_string = build_connection_string(args)
- logger.info(f"Connection string: {connection_string}")
+ try:
+ config = build_config(args)
+ except ValueError as error:
+ logger.error(f"Invalid client configuration: {error}")
+ return
logger.info(f"Connecting to {args.tcp_server_address} (TLS: {args.tls})")
- client = IggyClient.from_connection_string(connection_string)
+ client = IggyClient(config)
logger.info("Connecting to IggyClient")
+ # No login_user() call: auto_login replays the credentials on every
connect.
await client.connect()
logger.info("Connected.")
await init_system(client)
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index f709947b6..ff2cce834 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -47,4 +47,5 @@ pyo3-async-runtimes = { version = "0.29.0", features = [
"tokio-runtime",
] }
pyo3-stub-gen = "0.23.0"
+secrecy = "0.10"
tokio = "1.53.1"
diff --git a/foreign/python/README.md b/foreign/python/README.md
index 99a9b0ed6..8a5aff425 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md
@@ -139,6 +139,42 @@ running prek / committing / pushing. This list is not
exhaustive and other hook
./scripts/ci/markdownlint.sh --fix foreign/python/README.md # read the diff
after applying this, sometimes it gives unwanted results, e.g. messing up
enumerations
```
+## Client Configuration
+
+`IggyClient` takes either a server address or a `TcpConfig`:
+
+```python
+import asyncio
+from datetime import timedelta
+
+from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig
+
+
+async def main():
+ client = IggyClient(
+ TcpConfig(
+ server_address="127.0.0.1:8090",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ reconnection=TcpReconnectionConfig(
+ enabled=True,
+ max_retries=10,
+ interval=timedelta(seconds=2),
+ reestablish_after=timedelta(seconds=30),
+ ),
+ heartbeat_interval=timedelta(seconds=5),
+ # tls_enabled=True,
+ # tls_domain="localhost",
+ # tls_ca_file="../../core/certs/iggy_ca_cert.pem",
+ # tls_validate_certificate=True,
+ # nodelay=True,
+ )
+ )
+ await client.connect()
+
+
+asyncio.run(main())
+```
+
## Examples
Refer to the
[examples/python/](https://github.com/apache/iggy/tree/master/examples/python)
directory for usage examples.
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 341b3885c..ab63e66c5 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -29,6 +29,7 @@ __all__ = [
"AutoCommit",
"AutoCommitAfter",
"AutoCommitWhen",
+ "AutoLogin",
"ConsumerGroup",
"ConsumerGroupDetails",
"ConsumerGroupMember",
@@ -48,6 +49,8 @@ __all__ = [
"SendMessagesResponse",
"StreamDetails",
"StreamPermissions",
+ "TcpConfig",
+ "TcpReconnectionConfig",
"Topic",
"TopicDetails",
"TopicPermissions",
@@ -250,6 +253,41 @@ class AutoCommitWhen:
...
[email protected]
+class AutoLogin:
+ r"""
+ The credentials replayed by the client every time it (re)connects.
+
+ `IggyClient` only recovers a lost session when it has credentials to
replay,
+ so a long-running consumer should pass one of the enabled variants.
+ """
+ @property
+ def enabled(self) -> builtins.bool:
+ r"""
+ Whether automatic login is enabled.
+ """
+ @property
+ def username(self) -> builtins.str | None:
+ r"""
+ The username to log in with, or `None` for the disabled and token
variants.
+ """
+ @staticmethod
+ def disabled() -> AutoLogin:
+ r"""
+ No automatic login. `login_user()` must be called by hand after every
connect.
+ """
+ @staticmethod
+ def username_password(username: builtins.str, password: builtins.str) ->
AutoLogin:
+ r"""
+ Log in with the given username and password on every connect.
+ """
+ @staticmethod
+ def personal_access_token(token: builtins.str) -> AutoLogin:
+ r"""
+ Log in with the given personal access token on every connect.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class ConsumerGroup:
@property
@@ -772,14 +810,25 @@ class GlobalPermissions:
class IggyClient:
r"""
A Python class representing the Iggy client.
- It wraps the RustIggyClient and provides asynchronous functionality
- through the contained runtime.
+ It provides asynchronous functionality through the contained runtime.
"""
- def __new__(cls, conn: builtins.str | None = None) -> IggyClient:
+ def __new__(cls, conn: TcpConfig | builtins.str | None = None) ->
IggyClient:
r"""
- Constructs a new IggyClient from a TCP server address.
+ Constructs a new IggyClient from a TCP server address or a `TcpConfig`.
This initializes a new runtime for asynchronous operations.
Future versions might utilize asyncio for more Pythonic async.
+
+ Args:
+ conn: Either a `host:port` address, or a `TcpConfig` carrying the
full
+ transport configuration. Defaults to `127.0.0.1:8090` with
auto-login
+ disabled. A malformed address is reported differently by the
two
+ forms: the string form raises `RuntimeError` here, while
`TcpConfig`
+ raises `ValueError` when it is constructed, before it ever
reaches
+ this call. Neither exception is a subclass of the other.
+
+ Raises:
+ RuntimeError: If the address passed as a string is not a valid
+ `host:port` pair.
"""
@classmethod
def from_connection_string(cls, connection_string: builtins.str) ->
IggyClient:
@@ -790,15 +839,14 @@ class IggyClient:
def ping(self) -> collections.abc.Awaitable[None]:
r"""
Sends a ping request to the server to check connectivity.
- Returns `Ok(())` if the server responds successfully, or a
`PyRuntimeError`
- if the connection fails.
+ Raises `RuntimeError` if the connection fails.
"""
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.
+ Raises `RuntimeError` on failure.
"""
def get_user(
self, user_id: builtins.str | builtins.int
@@ -814,8 +862,8 @@ class IggyClient:
or `None` otherwise.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails.
"""
def get_users(self) -> collections.abc.Awaitable[list[UserInfo]]:
r"""
@@ -825,7 +873,7 @@ class IggyClient:
An awaitable that resolves to `list[UserInfo]`.
Raises:
- PyRuntimeError: If the request fails.
+ RuntimeError: If the request fails.
"""
def create_user(
self,
@@ -847,7 +895,7 @@ class IggyClient:
An awaitable that resolves to the created `UserInfoDetails`.
Raises:
- PyRuntimeError: If an argument is invalid or the request fails.
+ RuntimeError: If an argument is invalid or the request fails.
"""
def update_user(
self,
@@ -867,8 +915,8 @@ class IggyClient:
An awaitable that resolves to `None` when the user is updated.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails.
"""
def delete_user(
self, user_id: builtins.str | builtins.int
@@ -883,8 +931,8 @@ class IggyClient:
An awaitable that resolves to `None` when the user is deleted.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails.
"""
def update_permissions(
self, user_id: builtins.str | builtins.int, permissions: Permissions |
None
@@ -903,8 +951,8 @@ class IggyClient:
An awaitable that resolves to `None` when the permissions are
updated.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails.
"""
def change_password(
self,
@@ -924,8 +972,8 @@ class IggyClient:
An awaitable that resolves to `None` when the password is changed.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the current password is wrong or the request
fails.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the current password is wrong or the request
fails.
"""
def logout_user(self) -> collections.abc.Awaitable[None]:
r"""
@@ -935,24 +983,25 @@ class IggyClient:
An awaitable that resolves to `None` when the user is logged out.
Raises:
- PyRuntimeError: If the request fails.
+ RuntimeError: If the request fails.
"""
def connect(self) -> collections.abc.Awaitable[None]:
r"""
Connects the IggyClient to its service.
- Returns Ok(()) on successful connection or a PyRuntimeError on failure.
+ Raises `RuntimeError` if the connection fails.
"""
def create_stream(self, name: builtins.str) ->
collections.abc.Awaitable[None]:
r"""
Creates a new stream with the provided ID and name.
- Returns Ok(()) on successful stream creation or a PyRuntimeError on
failure.
+ Raises `RuntimeError` if the stream cannot be created.
"""
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.
+ Returns the stream details, or `None` if the stream does not exist.
+ Raises `RuntimeError` on failure.
"""
def create_topic(
self,
@@ -990,7 +1039,8 @@ class IggyClient:
) -> collections.abc.Awaitable[TopicDetails | None]:
r"""
Gets topic by stream and id.
- Returns Option of topic details or a PyRuntimeError on failure.
+ Returns the topic details, or `None` if the topic does not exist.
+ Raises `RuntimeError` on failure.
"""
def get_topics(
self, stream_id: builtins.str | builtins.int
@@ -1005,7 +1055,7 @@ class IggyClient:
An awaitable that resolves to `list[Topic]`.
Raises:
- PyRuntimeError: If the identifier is invalid or the request fails.
+ RuntimeError: If the identifier is invalid or the request fails.
"""
def update_topic(
self,
@@ -1055,7 +1105,7 @@ class IggyClient:
An awaitable that resolves to `None` when the topic is deleted.
Raises:
- PyRuntimeError: If an identifier is invalid or the request fails.
+ RuntimeError: If an identifier is invalid or the request fails.
"""
def purge_topic(
self,
@@ -1073,7 +1123,7 @@ class IggyClient:
An awaitable that resolves to `None` when the topic is purged.
Raises:
- PyRuntimeError: If an identifier is invalid or the request fails.
+ RuntimeError: If an identifier is invalid or the request fails.
"""
def create_consumer_group(
self,
@@ -1093,8 +1143,8 @@ class IggyClient:
An awaitable that resolves to `None` when the consumer group is
created.
Raises:
- PyValueError: If an identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If an identifier is invalid.
+ RuntimeError: If the request fails.
"""
def get_consumer_group(
self,
@@ -1115,8 +1165,8 @@ class IggyClient:
or `None` otherwise.
Raises:
- PyValueError: If an identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If an identifier is invalid.
+ RuntimeError: If the request fails.
"""
def get_consumer_groups(
self,
@@ -1134,8 +1184,8 @@ class IggyClient:
An awaitable that resolves to `list[ConsumerGroup]`.
Raises:
- PyValueError: If an identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If an identifier is invalid.
+ RuntimeError: If the request fails.
"""
def delete_consumer_group(
self,
@@ -1155,8 +1205,8 @@ class IggyClient:
An awaitable that resolves to `None` when the consumer group is
deleted.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails.
"""
def join_consumer_group(
self,
@@ -1179,8 +1229,8 @@ class IggyClient:
An awaitable that resolves to `None` when the client joins the
consumer group.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
"""
def leave_consumer_group(
self,
@@ -1205,8 +1255,8 @@ class IggyClient:
rejoin on their next poll.
Raises:
- PyValueError: If a string identifier is invalid.
- PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+ ValueError: If a string identifier is invalid.
+ RuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
"""
def send_messages(
self,
@@ -1233,7 +1283,7 @@ class IggyClient:
) -> 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.
+ Returns a list of received messages or a RuntimeError on failure.
"""
def consumer_group(
self,
@@ -1254,7 +1304,10 @@ class IggyClient:
) -> collections.abc.Awaitable[IggyConsumer]:
r"""
Creates a new consumer group consumer.
- Returns the consumer or a PyRuntimeError on failure.
+ Returns the consumer or a RuntimeError on failure. Raises `ValueError`
if
+ `poll_interval`, `polling_retry_interval`, `init_retry_interval` or an
+ `AutoCommit` interval is negative, or if any of those except
`poll_interval`
+ is zero.
"""
def send_binary_request(
self, code: builtins.int, payload: builtins.bytes
@@ -1273,15 +1326,14 @@ class IggyClient:
An awaitable that resolves to the raw response `bytes`.
Raises:
- PyRuntimeError: If the command cannot be sent or the server
returns an error.
+ RuntimeError: If the command cannot be sent or the server returns
an error.
"""
@typing.final
class IggyConsumer:
r"""
A Python class representing the Iggy consumer.
- It wraps the RustIggyConsumer and provides asynchronous functionality
- through the contained runtime.
+ It provides asynchronous functionality through the contained runtime.
"""
def get_last_consumed_offset(
self, partition_id: builtins.int
@@ -1315,8 +1367,7 @@ class IggyConsumer:
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.
+ Raises `RuntimeError` if the operation fails.
"""
def delete_offset(
self, partition_id: builtins.int | None
@@ -1324,14 +1375,13 @@ class IggyConsumer:
r"""
Deletes the 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.
+ Raises `RuntimeError` if the operation fails.
"""
def iter_messages(self) -> collections.abc.AsyncIterator[ReceiveMessage]:
r"""
Asynchronously iterate over `ReceiveMessage`s.
Returns an async iterator that raises `StopAsyncIteration` when no
more messages are available
- or a `PyRuntimeError` on failure.
+ or a `RuntimeError` on failure.
Note: This method does not currently support `AutoCommit.After`.
For `AutoCommit.IntervalOrAfter(datetime.timedelta, AutoCommitAfter)`,
only the interval part is applied; the `after` mode is ignored.
@@ -1346,7 +1396,7 @@ class IggyConsumer:
) -> 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.
+ Returns an awaitable that completes when shutdown is signaled or a
RuntimeError on failure.
"""
class IggyExpiry:
@@ -1546,7 +1596,7 @@ class PollingStrategy:
class ReceiveMessage:
r"""
A Python class representing a received message.
- This class wraps a Rust message, allowing for access to its payload and
offset from Python.
+ It provides access to the message payload and offset.
"""
def payload(self) -> bytes:
r"""
@@ -1596,8 +1646,6 @@ class ReceiveMessage:
class SendMessage:
r"""
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.
"""
def __new__(
cls,
@@ -1763,6 +1811,115 @@ class StreamPermissions:
treated as `None`.
"""
[email protected]
+class TcpConfig:
+ r"""
+ Configuration for the TCP transport, accepted by `IggyClient(...)`.
+
+ Every field is keyword-only and optional.
+ """
+ @property
+ def server_address(self) -> builtins.str: ...
+ @property
+ def auto_login(self) -> AutoLogin: ...
+ @property
+ def reconnection(self) -> TcpReconnectionConfig: ...
+ @property
+ def heartbeat_interval(self) -> datetime.timedelta: ...
+ @property
+ def tls_enabled(self) -> builtins.bool: ...
+ @property
+ def tls_domain(self) -> builtins.str: ...
+ @property
+ def tls_ca_file(self) -> builtins.str | None: ...
+ @property
+ def tls_validate_certificate(self) -> builtins.bool: ...
+ @property
+ def nodelay(self) -> builtins.bool: ...
+ def __new__(
+ cls,
+ *,
+ server_address: builtins.str | None = None,
+ auto_login: AutoLogin | None = None,
+ reconnection: TcpReconnectionConfig | None = None,
+ heartbeat_interval: datetime.timedelta | None = None,
+ tls_enabled: builtins.bool | None = None,
+ tls_domain: builtins.str | None = None,
+ tls_ca_file: builtins.str | None = None,
+ tls_validate_certificate: builtins.bool | None = None,
+ nodelay: builtins.bool | None = None,
+ ) -> TcpConfig:
+ r"""
+ Constructs a TCP configuration.
+
+ Args:
+ server_address: `host:port` of the Iggy server. Defaults to
`127.0.0.1:8090`.
+ auto_login: Credentials replayed on every connect. Defaults to
`AutoLogin.disabled()`.
+ reconnection: Reconnection policy. Defaults to
`TcpReconnectionConfig()`.
+ heartbeat_interval: Interval of heartbeats sent by the client.
Defaults to 5 seconds.
+ tls_enabled: Whether to connect over TLS. Defaults to disabled.
+ tls_domain: Domain to validate the certificate against. Empty
means it is
+ taken from `server_address`.
+ tls_ca_file: Path to the CA file for TLS. Read only when
`tls_enabled`
+ and `tls_validate_certificate` are both on; with either one
off it
+ is kept but never consulted, so pairing it with
+ `tls_validate_certificate=False` pins nothing.
+ tls_validate_certificate: Whether to validate the server
certificate.
+ Defaults to validating. Disabling this accepts any certificate
the
+ server presents, including self-signed and mismatched ones, and
+ takes precedence over `tls_ca_file`; intended for local
development
+ only.
+ nodelay: Disable the Nagle algorithm for the TCP socket. Defaults
to
+ leaving it on.
+
+ Raises:
+ ValueError: If `server_address` is not a valid `host:port` pair,
if a
+ duration is negative, or if `heartbeat_interval` is zero.
+ """
+ def __repr__(self) -> builtins.str: ...
+
[email protected]
+class TcpReconnectionConfig:
+ r"""
+ How the TCP client reconnects after the connection to the server is lost.
+ """
+ @property
+ def enabled(self) -> builtins.bool: ...
+ @property
+ def max_retries(self) -> builtins.int | None: ...
+ @property
+ def interval(self) -> datetime.timedelta: ...
+ @property
+ def reestablish_after(self) -> datetime.timedelta: ...
+ def __new__(
+ cls,
+ *,
+ enabled: builtins.bool | None = None,
+ max_retries: builtins.int | None = None,
+ interval: datetime.timedelta | None = None,
+ reestablish_after: datetime.timedelta | None = None,
+ ) -> TcpReconnectionConfig:
+ r"""
+ Constructs a reconnection policy.
+
+ Args:
+ enabled: Whether to reconnect at all. Defaults to enabled.
+ max_retries: Attempts before giving up, or `None` for unlimited.
+ Defaults to unlimited, which means a call awaited while the
server
+ is down never returns: `connect()`, `send_messages()` and
+ `poll_messages()` all wait inside the retry loop. Set a finite
+ number for request/reply style usage, so a call fails instead.
+ interval: Delay between attempts. Defaults to 1 second.
+ reestablish_after: Cooldown before reconnecting after a previously
+ successful connection. Defaults to 5 seconds.
+
+ Raises:
+ ValueError: If a duration is negative, if `max_retries` is outside
the
+ range of an unsigned 32-bit integer, or if `interval` is zero
while
+ reconnection is enabled and `max_retries` is unlimited.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class Topic:
@property
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index c7cbe3952..cde56040f 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -30,10 +30,12 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass,
gen_stub_pymethods};
use std::str::FromStr;
use std::sync::Arc;
+use crate::config::PyClientConfig;
use crate::consumer::{
AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as
PyConsumerGroupDetails,
- IggyConsumer, py_delta_to_iggy_duration,
+ IggyConsumer,
};
+use crate::duration::{py_delta_to_iggy_duration, reject_zero};
use crate::identifier::PyIdentifier;
use crate::permissions::Permissions as PyPermissions;
use crate::receive_message::{PollingStrategy, ReceiveMessage};
@@ -46,8 +48,7 @@ use crate::user::{
use tokio::sync::Mutex;
/// A Python class representing the Iggy client.
-/// It wraps the RustIggyClient and provides asynchronous functionality
-/// through the contained runtime.
+/// It provides asynchronous functionality through the contained runtime.
#[gen_stub_pyclass]
#[pyclass]
pub struct IggyClient {
@@ -83,21 +84,44 @@ fn resolve_topic_params(
#[gen_stub_pymethods]
#[pymethods]
impl IggyClient {
- /// Constructs a new IggyClient from a TCP server address.
+ /// Constructs a new IggyClient from a TCP server address or a `TcpConfig`.
/// This initializes a new runtime for asynchronous operations.
/// Future versions might utilize asyncio for more Pythonic async.
+ ///
+ /// Args:
+ /// conn: Either a `host:port` address, or a `TcpConfig` carrying the
full
+ /// transport configuration. Defaults to `127.0.0.1:8090` with
auto-login
+ /// disabled. A malformed address is reported differently by the
two
+ /// forms: the string form raises `RuntimeError` here, while
`TcpConfig`
+ /// raises `ValueError` when it is constructed, before it ever
reaches
+ /// this call. Neither exception is a subclass of the other.
+ ///
+ /// Raises:
+ /// RuntimeError: If the address passed as a string is not a valid
+ /// `host:port` pair.
#[new]
#[pyo3(signature = (conn=None))]
fn new(
- #[gen_stub(override_type(type_repr = "builtins.str | None"))] conn:
Option<String>,
+ #[gen_stub(override_type(type_repr = "TcpConfig | builtins.str |
None"))] conn: Option<
+ PyClientConfig,
+ >,
) -> PyResult<Self> {
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(conn.unwrap_or("127.0.0.1:8090".to_string()))
- .build()
+ let config = match conn {
+ Some(PyClientConfig::Config(config)) => config.client_config(),
+ Some(PyClientConfig::ServerAddress(server_address)) => Arc::new(
+ TcpClientConfigBuilder::new()
+ .with_server_address(server_address)
+ .build()
+ .map_err(|e| {
+ PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string())
+ })?,
+ ),
+ None => Arc::new(TcpClientConfig::default()),
+ };
+ let tcp_client = TcpClient::create(config)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?;
Ok(IggyClient {
- inner: Arc::new(client),
+ inner:
Arc::new(RustIggyClient::new(ClientWrapper::Tcp(tcp_client))),
})
}
@@ -119,8 +143,7 @@ impl IggyClient {
}
/// Sends a ping request to the server to check connectivity.
- /// Returns `Ok(())` if the server responds successfully, or a
`PyRuntimeError`
- /// if the connection fails.
+ /// Raises `RuntimeError` if the connection fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn ping<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
let inner = self.inner.clone();
@@ -133,7 +156,7 @@ impl IggyClient {
}
/// Logs in the user with the given credentials.
- /// Returns `Ok(())` on success, or a PyRuntimeError on failure.
+ /// Raises `RuntimeError` on failure.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn login_user<'a>(
&self,
@@ -161,8 +184,8 @@ impl IggyClient {
/// or `None` otherwise.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[UserInfoDetails
| None]", imports=("collections.abc")))]
fn get_user<'a>(&self, py: Python<'a>, user_id: PyIdentifier) ->
PyResult<Bound<'a, PyAny>> {
let user_id = Identifier::try_from(user_id)?;
@@ -183,7 +206,7 @@ impl IggyClient {
/// An awaitable that resolves to `list[UserInfo]`.
///
/// Raises:
- /// PyRuntimeError: If the request fails.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[UserInfo]]",
imports=("collections.abc")))]
fn get_users<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
let inner = self.inner.clone();
@@ -209,7 +232,7 @@ impl IggyClient {
/// An awaitable that resolves to the created `UserInfoDetails`.
///
/// Raises:
- /// PyRuntimeError: If an argument is invalid or the request fails.
+ /// RuntimeError: If an argument is invalid or the request fails.
#[pyo3(signature = (username, password, status=None, permissions=None))]
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[UserInfoDetails]",
imports=("collections.abc")))]
fn create_user<'a>(
@@ -246,8 +269,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the user is updated.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails.
#[pyo3(signature = (user_id, username=None, status=None))]
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn update_user<'a>(
@@ -279,8 +302,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the user is deleted.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn delete_user<'a>(&self, py: Python<'a>, user_id: PyIdentifier) ->
PyResult<Bound<'a, PyAny>> {
let user_id = Identifier::try_from(user_id)?;
@@ -308,8 +331,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the permissions are
updated.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails.
#[pyo3(signature = (user_id, permissions))]
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn update_permissions<'a>(
@@ -344,8 +367,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the password is changed.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the current password is wrong or the request
fails.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the current password is wrong or the request
fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn change_password<'a>(
&self,
@@ -372,7 +395,7 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the user is logged out.
///
/// Raises:
- /// PyRuntimeError: If the request fails.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn logout_user<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
let inner = self.inner.clone();
@@ -387,7 +410,7 @@ impl IggyClient {
}
/// Connects the IggyClient to its service.
- /// Returns Ok(()) on successful connection or a PyRuntimeError on failure.
+ /// Raises `RuntimeError` if the connection fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn connect<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
let inner = self.inner.clone();
@@ -401,7 +424,7 @@ impl IggyClient {
}
/// Creates a new stream with the provided ID and name.
- /// Returns Ok(()) on successful stream creation or a PyRuntimeError on
failure.
+ /// Raises `RuntimeError` if the stream cannot be created.
#[pyo3(signature = (name))]
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn create_stream<'a>(&self, py: Python<'a>, name: String) ->
PyResult<Bound<'a, PyAny>> {
@@ -416,7 +439,8 @@ impl IggyClient {
}
/// Gets stream by id.
- /// Returns Option of stream details or a PyRuntimeError on failure.
+ /// Returns the stream details, or `None` if the stream does not exist.
+ /// Raises `RuntimeError` on failure.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[StreamDetails
| None]", imports=("collections.abc")))]
fn get_stream<'a>(
&self,
@@ -500,7 +524,8 @@ impl IggyClient {
}
/// Gets topic by stream and id.
- /// Returns Option of topic details or a PyRuntimeError on failure.
+ /// Returns the topic details, or `None` if the topic does not exist.
+ /// Raises `RuntimeError` on failure.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[TopicDetails
| None]", imports=("collections.abc")))]
fn get_topic<'a>(
&self,
@@ -530,7 +555,7 @@ impl IggyClient {
/// An awaitable that resolves to `list[Topic]`.
///
/// Raises:
- /// PyRuntimeError: If the identifier is invalid or the request fails.
+ /// RuntimeError: If the identifier is invalid or the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[Topic]]",
imports=("collections.abc")))]
fn get_topics<'a>(
&self,
@@ -627,7 +652,7 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the topic is deleted.
///
/// Raises:
- /// PyRuntimeError: If an identifier is invalid or the request fails.
+ /// RuntimeError: If an identifier is invalid or the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn delete_topic<'a>(
&self,
@@ -658,7 +683,7 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the topic is purged.
///
/// Raises:
- /// PyRuntimeError: If an identifier is invalid or the request fails.
+ /// RuntimeError: If an identifier is invalid or the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn purge_topic<'a>(
&self,
@@ -690,8 +715,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the consumer group is
created.
///
/// Raises:
- /// PyValueError: If an identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If an identifier is invalid.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn create_consumer_group<'a>(
&self,
@@ -725,8 +750,8 @@ impl IggyClient {
/// or `None` otherwise.
///
/// Raises:
- /// PyValueError: If an identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If an identifier is invalid.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[ConsumerGroupDetails
| None]", imports=("collections.abc")))]
fn get_consumer_group<'a>(
&self,
@@ -759,8 +784,8 @@ impl IggyClient {
/// An awaitable that resolves to `list[ConsumerGroup]`.
///
/// Raises:
- /// PyValueError: If an identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If an identifier is invalid.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[ConsumerGroup]]",
imports=("collections.abc")))]
fn get_consumer_groups<'a>(
&self,
@@ -795,8 +820,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the consumer group is
deleted.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn delete_consumer_group<'a>(
&self,
@@ -833,8 +858,8 @@ impl IggyClient {
/// An awaitable that resolves to `None` when the client joins the
consumer group.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn join_consumer_group<'a>(
&self,
@@ -873,8 +898,8 @@ impl IggyClient {
/// rejoin on their next poll.
///
/// Raises:
- /// PyValueError: If a string identifier is invalid.
- /// PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+ /// ValueError: If a string identifier is invalid.
+ /// RuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn leave_consumer_group<'a>(
&self,
@@ -938,7 +963,7 @@ impl IggyClient {
}
/// Polls for messages from the specified topic and partition.
- /// Returns a list of received messages or a PyRuntimeError on failure.
+ /// Returns a list of received messages or a RuntimeError on failure.
#[allow(clippy::too_many_arguments)]
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[ReceiveMessage]]",
imports=("collections.abc")))]
fn poll_messages<'a>(
@@ -984,7 +1009,10 @@ impl IggyClient {
}
/// Creates a new consumer group consumer.
- /// Returns the consumer or a PyRuntimeError on failure.
+ /// Returns the consumer or a RuntimeError on failure. Raises `ValueError`
if
+ /// `poll_interval`, `polling_retry_interval`, `init_retry_interval` or an
+ /// `AutoCommit` interval is negative, or if any of those except
`poll_interval`
+ /// is zero.
#[allow(clippy::too_many_arguments)]
#[pyo3(signature = (
name,
@@ -1060,8 +1088,10 @@ impl IggyClient {
builder = builder.without_poll_interval()
};
if let Some(polling_retry_interval) = polling_retry_interval {
- builder =
-
builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval)?)
+ builder = builder.polling_retry_interval(reject_zero(
+ py_delta_to_iggy_duration(&polling_retry_interval)?,
+ "polling_retry_interval",
+ )?)
}
if init_retries.is_some() && init_retry_interval.is_none() {
return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
@@ -1077,7 +1107,10 @@ impl IggyClient {
{
builder = builder.init_retries(
init_retries,
- py_delta_to_iggy_duration(&init_retry_interval)?,
+ reject_zero(
+ py_delta_to_iggy_duration(&init_retry_interval)?,
+ "init_retry_interval",
+ )?,
);
}
if allow_replay {
@@ -1109,7 +1142,7 @@ impl IggyClient {
/// An awaitable that resolves to the raw response `bytes`.
///
/// Raises:
- /// PyRuntimeError: If the command cannot be sent or the server
returns an error.
+ /// RuntimeError: If the command cannot be sent or the server returns
an error.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[bytes]",
imports=("collections.abc")))]
fn send_binary_request<'a>(
&self,
diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs
new file mode 100644
index 000000000..4519c939f
--- /dev/null
+++ b/foreign/python/src/config.rs
@@ -0,0 +1,423 @@
+// 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::{
+ AutoLogin as RustAutoLogin, Credentials as RustCredentials,
+ TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder,
+ TcpClientReconnectionConfig as RustTcpClientReconnectionConfig,
+};
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+use pyo3::types::PyDelta;
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
+use pyo3_stub_gen::impl_stub_type;
+use secrecy::SecretString;
+use std::sync::Arc;
+
+use crate::duration::{
+ duration_repr, iggy_duration_to_py_delta, py_delta_to_iggy_duration,
reject_zero,
+};
+
+/// The credentials replayed by the client every time it (re)connects.
+///
+/// `IggyClient` only recovers a lost session when it has credentials to
replay,
+/// so a long-running consumer should pass one of the enabled variants.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct AutoLogin {
+ pub(crate) inner: RustAutoLogin,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl AutoLogin {
+ /// No automatic login. `login_user()` must be called by hand after every
connect.
+ #[staticmethod]
+ fn disabled() -> Self {
+ Self {
+ inner: RustAutoLogin::Disabled,
+ }
+ }
+
+ /// Log in with the given username and password on every connect.
+ #[staticmethod]
+ fn username_password(username: String, password: String) -> Self {
+ Self {
+ inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword(
+ username,
+ SecretString::from(password),
+ )),
+ }
+ }
+
+ /// Log in with the given personal access token on every connect.
+ #[staticmethod]
+ fn personal_access_token(token: String) -> Self {
+ Self {
+ inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(
+ SecretString::from(token),
+ )),
+ }
+ }
+
+ /// Whether automatic login is enabled.
+ #[getter]
+ fn enabled(&self) -> bool {
+ matches!(self.inner, RustAutoLogin::Enabled(_))
+ }
+
+ /// The username to log in with, or `None` for the disabled and token
variants.
+ #[gen_stub(override_return_type(type_repr = "builtins.str | None"))]
+ #[getter]
+ fn username(&self) -> Option<String> {
+ match &self.inner {
+ RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username,
_)) => {
+ Some(username.clone())
+ }
+ _ => None,
+ }
+ }
+
+ fn __repr__(&self) -> String {
+ match &self.inner {
+ RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(),
+ RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username,
_)) => {
+ format!("AutoLogin.username_password({username:?}, ...)")
+ }
+ RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) =>
{
+ "AutoLogin.personal_access_token(...)".to_owned()
+ }
+ }
+ }
+}
+
+/// How the TCP client reconnects after the connection to the server is lost.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct TcpReconnectionConfig {
+ pub(crate) inner: RustTcpClientReconnectionConfig,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl TcpReconnectionConfig {
+ /// Constructs a reconnection policy.
+ ///
+ /// Args:
+ /// enabled: Whether to reconnect at all. Defaults to enabled.
+ /// max_retries: Attempts before giving up, or `None` for unlimited.
+ /// Defaults to unlimited, which means a call awaited while the
server
+ /// is down never returns: `connect()`, `send_messages()` and
+ /// `poll_messages()` all wait inside the retry loop. Set a finite
+ /// number for request/reply style usage, so a call fails instead.
+ /// interval: Delay between attempts. Defaults to 1 second.
+ /// reestablish_after: Cooldown before reconnecting after a previously
+ /// successful connection. Defaults to 5 seconds.
+ ///
+ /// Raises:
+ /// ValueError: If a duration is negative, if `max_retries` is outside
the
+ /// range of an unsigned 32-bit integer, or if `interval` is zero
while
+ /// reconnection is enabled and `max_retries` is unlimited.
+ #[new]
+ #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None,
reestablish_after=None))]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
enabled: Option<bool>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
max_retries: Option<i64>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ reestablish_after: Option<Py<PyDelta>>,
+ ) -> PyResult<Self> {
+ let defaults = RustTcpClientReconnectionConfig::default();
+ let enabled = enabled.unwrap_or(defaults.enabled);
+ let max_retries = max_retries
+ .map(|max_retries| {
+ u32::try_from(max_retries).map_err(|_| {
+ PyValueError::new_err(format!(
+ "'max_retries' must be between 0 and {}",
+ u32::MAX
+ ))
+ })
+ })
+ .transpose()?;
+ let interval = interval
+ .as_ref()
+ .map(py_delta_to_iggy_duration)
+ .transpose()?
+ .unwrap_or(defaults.interval);
+ // Unlimited retries at a zero interval reconnect in a continuous loop;
+ // a zero interval with a retry cap is a legitimate fast-retry policy,
and
+ // with reconnection off the interval is never read at all.
+ if enabled && interval.is_zero() && max_retries.is_none() {
+ return Err(PyValueError::new_err(
+ "'interval' must not be zero unless 'max_retries' is set",
+ ));
+ }
+ Ok(Self {
+ inner: RustTcpClientReconnectionConfig {
+ enabled,
+ max_retries,
+ interval,
+ reestablish_after: reestablish_after
+ .as_ref()
+ .map(py_delta_to_iggy_duration)
+ .transpose()?
+ .unwrap_or(defaults.reestablish_after),
+ },
+ })
+ }
+
+ #[getter]
+ fn enabled(&self) -> bool {
+ self.inner.enabled
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn max_retries(&self) -> Option<u32> {
+ self.inner.max_retries
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ #[getter]
+ fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.interval)
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ #[getter]
+ fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a,
PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.reestablish_after)
+ }
+
+ fn __repr__(&self) -> String {
+ let max_retries = match self.inner.max_retries {
+ Some(max_retries) => max_retries.to_string(),
+ None => "None".to_owned(),
+ };
+ format!(
+ "TcpReconnectionConfig(enabled={}, max_retries={max_retries},
interval={}, reestablish_after={})",
+ python_bool(self.inner.enabled),
+ duration_repr(self.inner.interval),
+ duration_repr(self.inner.reestablish_after),
+ )
+ }
+}
+
+/// Configuration for the TCP transport, accepted by `IggyClient(...)`.
+///
+/// Every field is keyword-only and optional.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct TcpConfig {
+ inner: Arc<RustTcpClientConfig>,
+}
+
+impl TcpConfig {
+ /// The configuration in the shape `TcpClient::create` expects.
+ pub(crate) fn client_config(&self) -> Arc<RustTcpClientConfig> {
+ self.inner.clone()
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl TcpConfig {
+ /// Constructs a TCP configuration.
+ ///
+ /// Args:
+ /// server_address: `host:port` of the Iggy server. Defaults to
`127.0.0.1:8090`.
+ /// auto_login: Credentials replayed on every connect. Defaults to
`AutoLogin.disabled()`.
+ /// reconnection: Reconnection policy. Defaults to
`TcpReconnectionConfig()`.
+ /// heartbeat_interval: Interval of heartbeats sent by the client.
Defaults to 5 seconds.
+ /// tls_enabled: Whether to connect over TLS. Defaults to disabled.
+ /// tls_domain: Domain to validate the certificate against. Empty
means it is
+ /// taken from `server_address`.
+ /// tls_ca_file: Path to the CA file for TLS. Read only when
`tls_enabled`
+ /// and `tls_validate_certificate` are both on; with either one
off it
+ /// is kept but never consulted, so pairing it with
+ /// `tls_validate_certificate=False` pins nothing.
+ /// tls_validate_certificate: Whether to validate the server
certificate.
+ /// Defaults to validating. Disabling this accepts any certificate
the
+ /// server presents, including self-signed and mismatched ones, and
+ /// takes precedence over `tls_ca_file`; intended for local
development
+ /// only.
+ /// nodelay: Disable the Nagle algorithm for the TCP socket. Defaults
to
+ /// leaving it on.
+ ///
+ /// Raises:
+ /// ValueError: If `server_address` is not a valid `host:port` pair,
if a
+ /// duration is negative, or if `heartbeat_interval` is zero.
+ #[new]
+ #[pyo3(signature = (
+ *,
+ server_address=None,
+ auto_login=None,
+ reconnection=None,
+ heartbeat_interval=None,
+ tls_enabled=None,
+ tls_domain=None,
+ tls_ca_file=None,
+ tls_validate_certificate=None,
+ nodelay=None,
+ ))]
+ #[allow(clippy::too_many_arguments)]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))]
server_address: Option<
+ String,
+ >,
+ #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login:
Option<AutoLogin>,
+ #[gen_stub(override_type(type_repr = "TcpReconnectionConfig | None"))]
reconnection: Option<
+ TcpReconnectionConfig,
+ >,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ heartbeat_interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
tls_enabled: Option<bool>,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))]
tls_domain: Option<String>,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))]
tls_ca_file: Option<String>,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
+ tls_validate_certificate: Option<bool>,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
nodelay: Option<bool>,
+ ) -> PyResult<Self> {
+ // The builder starts from `TcpClientConfig::default()`, and its
`build()`
+ // trims and validates the address whether or not one was set here.
+ let mut builder = TcpClientConfigBuilder::new();
+ if let Some(server_address) = server_address {
+ builder = builder.with_server_address(server_address);
+ }
+ let mut inner = builder
+ .build()
+ .map_err(|e| PyValueError::new_err(e.to_string()))?;
+ if let Some(auto_login) = auto_login {
+ inner.auto_login = auto_login.inner;
+ }
+ if let Some(reconnection) = reconnection {
+ inner.reconnection = reconnection.inner;
+ }
+ if let Some(heartbeat_interval) = heartbeat_interval {
+ inner.heartbeat_interval = reject_zero(
+ py_delta_to_iggy_duration(&heartbeat_interval)?,
+ "heartbeat_interval",
+ )?;
+ }
+ if let Some(tls_enabled) = tls_enabled {
+ inner.tls_enabled = tls_enabled;
+ }
+ if let Some(tls_domain) = tls_domain {
+ inner.tls_domain = tls_domain;
+ }
+ if tls_ca_file.is_some() {
+ inner.tls_ca_file = tls_ca_file;
+ }
+ if let Some(tls_validate_certificate) = tls_validate_certificate {
+ inner.tls_validate_certificate = tls_validate_certificate;
+ }
+ if let Some(nodelay) = nodelay {
+ inner.nodelay = nodelay;
+ }
+
+ Ok(Self {
+ inner: Arc::new(inner),
+ })
+ }
+
+ #[getter]
+ fn server_address(&self) -> String {
+ self.inner.server_address.clone()
+ }
+
+ #[getter]
+ fn auto_login(&self) -> AutoLogin {
+ AutoLogin {
+ inner: self.inner.auto_login.clone(),
+ }
+ }
+
+ #[getter]
+ fn reconnection(&self) -> TcpReconnectionConfig {
+ TcpReconnectionConfig {
+ inner: self.inner.reconnection.clone(),
+ }
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ #[getter]
+ fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a,
PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.heartbeat_interval)
+ }
+
+ #[getter]
+ fn tls_enabled(&self) -> bool {
+ self.inner.tls_enabled
+ }
+
+ #[getter]
+ fn tls_domain(&self) -> String {
+ self.inner.tls_domain.clone()
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.str | None"))]
+ #[getter]
+ fn tls_ca_file(&self) -> Option<String> {
+ self.inner.tls_ca_file.clone()
+ }
+
+ #[getter]
+ fn tls_validate_certificate(&self) -> bool {
+ self.inner.tls_validate_certificate
+ }
+
+ #[getter]
+ fn nodelay(&self) -> bool {
+ self.inner.nodelay
+ }
+
+ fn __repr__(&self) -> String {
+ let tls_ca_file = match &self.inner.tls_ca_file {
+ Some(tls_ca_file) => format!("{tls_ca_file:?}"),
+ None => "None".to_owned(),
+ };
+ format!(
+ "TcpConfig(server_address={:?}, auto_login={}, reconnection={},
heartbeat_interval={}, tls_enabled={}, tls_domain={:?},
tls_ca_file={tls_ca_file}, tls_validate_certificate={}, nodelay={})",
+ self.inner.server_address,
+ self.auto_login().__repr__(),
+ self.reconnection().__repr__(),
+ duration_repr(self.inner.heartbeat_interval),
+ python_bool(self.inner.tls_enabled),
+ self.inner.tls_domain,
+ python_bool(self.inner.tls_validate_certificate),
+ python_bool(self.inner.nodelay),
+ )
+ }
+}
+
+fn python_bool(value: bool) -> &'static str {
+ if value { "True" } else { "False" }
+}
+
+/// What `IggyClient(...)` accepts: a bare `host:port` or a full `TcpConfig`.
+#[derive(FromPyObject)]
+pub enum PyClientConfig {
+ #[pyo3(transparent)]
+ Config(TcpConfig),
+ #[pyo3(transparent, annotation = "str")]
+ ServerAddress(String),
+}
+impl_stub_type!(PyClientConfig = TcpConfig | String);
diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs
index 4d64fc626..6a6e69a87 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -16,7 +16,6 @@
// under the License.
use std::sync::Arc;
-use std::time::Duration;
use futures::StreamExt;
use iggy::consumer_ext::{IggyConsumerMessageExt, MessageConsumer};
@@ -27,8 +26,8 @@ use iggy::prelude::{
ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as
RustIggyConsumer, IggyDuration,
IggyError, ReceivedMessage,
};
-use pyo3::exceptions::{PyStopAsyncIteration, PyValueError};
-use pyo3::types::{PyDelta, PyDeltaAccess};
+use pyo3::exceptions::PyStopAsyncIteration;
+use pyo3::types::PyDelta;
use pyo3::prelude::*;
use pyo3_async_runtimes::TaskLocals;
@@ -39,12 +38,12 @@ use tokio::sync::Mutex;
use tokio::sync::oneshot::Sender;
use tokio::task::JoinHandle;
+use crate::duration::{py_delta_to_iggy_duration, reject_zero};
use crate::identifier::PyIdentifier;
use crate::receive_message::ReceiveMessage;
/// A Python class representing the Iggy consumer.
-/// It wraps the RustIggyConsumer and provides asynchronous functionality
-/// through the contained runtime.
+/// It provides asynchronous functionality through the contained runtime.
#[gen_stub_pyclass]
#[pyclass]
pub struct IggyConsumer {
@@ -94,8 +93,7 @@ impl IggyConsumer {
/// 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.
+ /// Raises `RuntimeError` if the operation fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn store_offset<'a>(
&self,
@@ -116,8 +114,7 @@ impl IggyConsumer {
/// Deletes the 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.
+ /// Raises `RuntimeError` if the operation fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn delete_offset<'a>(
&self,
@@ -137,7 +134,7 @@ impl IggyConsumer {
/// Asynchronously iterate over `ReceiveMessage`s.
/// Returns an async iterator that raises `StopAsyncIteration` when no
more messages are available
- /// or a `PyRuntimeError` on failure.
+ /// or a `RuntimeError` on failure.
/// Note: This method does not currently support `AutoCommit.After`.
/// For `AutoCommit.IntervalOrAfter(datetime.timedelta, AutoCommitAfter)`,
/// only the interval part is applied; the `after` mode is ignored.
@@ -149,7 +146,7 @@ impl IggyConsumer {
}
/// 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.
+ /// Returns an awaitable that completes when shutdown is signaled or a
RuntimeError on failure.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn consume_messages<'a>(
&self,
@@ -435,17 +432,12 @@ impl TryFrom<&AutoCommit> for RustAutoCommit {
fn try_from(val: &AutoCommit) -> PyResult<RustAutoCommit> {
Ok(match val {
AutoCommit::Disabled() => RustAutoCommit::Disabled,
- AutoCommit::Interval(delta) => {
- let duration = py_delta_to_iggy_duration(delta)?;
- RustAutoCommit::Interval(duration)
- }
+ AutoCommit::Interval(delta) =>
RustAutoCommit::Interval(auto_commit_interval(delta)?),
AutoCommit::IntervalOrWhen(delta, when) => {
- let duration = py_delta_to_iggy_duration(delta)?;
- RustAutoCommit::IntervalOrWhen(duration, when.into())
+ RustAutoCommit::IntervalOrWhen(auto_commit_interval(delta)?,
when.into())
}
AutoCommit::IntervalOrAfter(delta, after) => {
- let duration = py_delta_to_iggy_duration(delta)?;
- RustAutoCommit::IntervalOrAfter(duration, after.into())
+ RustAutoCommit::IntervalOrAfter(auto_commit_interval(delta)?,
after.into())
}
AutoCommit::When(when) => RustAutoCommit::When(when.into()),
AutoCommit::After(after) => RustAutoCommit::After(after.into()),
@@ -453,6 +445,10 @@ impl TryFrom<&AutoCommit> for RustAutoCommit {
}
}
+fn auto_commit_interval(delta: &Py<PyDelta>) -> PyResult<IggyDuration> {
+ reject_zero(py_delta_to_iggy_duration(delta)?, "AutoCommit interval")
+}
+
/// The auto-commit mode for storing the offset on the server.
#[derive(Debug, PartialEq, Copy, Clone)]
#[gen_stub_pyclass_complex_enum(skip_stub_type)]
@@ -518,20 +514,3 @@ impl PyStubType for AutoCommitAfter {
TypeInfo::unqualified("AutoCommitAfter")
}
}
-
-pub fn py_delta_to_iggy_duration(delta1: &Py<PyDelta>) ->
PyResult<IggyDuration> {
- Python::attach(|py| {
- let delta = delta1.bind(py);
- let total_seconds = i64::from(delta.get_days()) * 86_400 +
i64::from(delta.get_seconds());
- if total_seconds < 0 {
- return Err(PyValueError::new_err(
- "duration must not be negative".to_string(),
- ));
- }
- let nanos = (delta.get_microseconds() * 1_000) as u32;
- Ok(IggyDuration::new(Duration::new(
- total_seconds as u64,
- nanos,
- )))
- })
-}
diff --git a/foreign/python/src/duration.rs b/foreign/python/src/duration.rs
new file mode 100644
index 000000000..9b2b419fe
--- /dev/null
+++ b/foreign/python/src/duration.rs
@@ -0,0 +1,65 @@
+// 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::IggyDuration;
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+use pyo3::types::PyDelta;
+use std::time::Duration;
+
+pub fn py_delta_to_iggy_duration(delta: &Py<PyDelta>) ->
PyResult<IggyDuration> {
+ Python::attach(|py| {
+ // The value is already a timedelta, so a negative one is the only
failure
+ // left to map, and the Python surface must not name Rust types.
+ delta
+ .bind(py)
+ .extract::<Duration>()
+ .map(IggyDuration::from)
+ .map_err(|_| PyValueError::new_err("duration must not be
negative"))
+ })
+}
+
+pub fn iggy_duration_to_py_delta(
+ py: Python<'_>,
+ duration: IggyDuration,
+) -> PyResult<Bound<'_, PyDelta>> {
+ duration.get_duration().into_pyobject(py)
+}
+
+/// Renders a duration the way it would be written in Python, so that a
`__repr__`
+/// built from it can be pasted back into a constructor.
+pub fn duration_repr(duration: IggyDuration) -> String {
+ // Read the std duration, whose micros are u128:
`IggyDuration::as_micros()`
+ // truncates to u64, which a timedelta near the Python maximum overflows.
+ let micros = duration.get_duration().as_micros();
+ if micros.is_multiple_of(1_000_000) {
+ format!("datetime.timedelta(seconds={})", micros / 1_000_000)
+ } else {
+ format!("datetime.timedelta(microseconds={micros})")
+ }
+}
+
+/// Rejects a zero duration for parameters where zero means an unthrottled loop
+/// rather than "disabled".
+pub fn reject_zero(duration: IggyDuration, parameter: &str) ->
PyResult<IggyDuration> {
+ if duration.is_zero() {
+ return Err(PyValueError::new_err(format!(
+ "'{parameter}' must not be zero"
+ )));
+ }
+ Ok(duration)
+}
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 985476ff7..9c7f1efaa 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -16,7 +16,9 @@
// under the License.
pub mod client;
+mod config;
mod consumer;
+mod duration;
mod identifier;
mod permissions;
mod receive_message;
@@ -27,6 +29,7 @@ mod user;
mod user_headers;
use client::IggyClient;
+use config::{AutoLogin, TcpConfig, TcpReconnectionConfig};
use consumer::{
AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup,
ConsumerGroupDetails,
ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator,
@@ -40,7 +43,7 @@ use topic::{IggyExpiry, MaxTopicSize, Partition, Topic,
TopicDetails};
use user::{UserInfo, UserInfoDetails, UserStatus};
use user_headers::{HeaderKey, HeaderValue, UserHeaders};
-/// A Python module implemented in Rust.
+/// Python client for Apache Iggy, the persistent message streaming platform.
#[pymodule]
fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<SendMessage>()?;
@@ -48,6 +51,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) ->
PyResult<()> {
m.add_class::<SendMessagesConfirmation>()?;
m.add_class::<ReceiveMessage>()?;
m.add_class::<IggyClient>()?;
+ m.add_class::<AutoLogin>()?;
+ m.add_class::<TcpConfig>()?;
+ m.add_class::<TcpReconnectionConfig>()?;
m.add_class::<StreamDetails>()?;
m.add_class::<Topic>()?;
m.add_class::<TopicDetails>()?;
diff --git a/foreign/python/src/receive_message.rs
b/foreign/python/src/receive_message.rs
index ecabb6cf2..ddb14e623 100644
--- a/foreign/python/src/receive_message.rs
+++ b/foreign/python/src/receive_message.rs
@@ -24,7 +24,7 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass,
gen_stub_pyclass_complex_enum, gen
use crate::user_headers::{UserHeaders, rust_user_headers_to_py};
/// A Python class representing a received message.
-/// This class wraps a Rust message, allowing for access to its payload and
offset from Python.
+/// It provides access to the message payload and offset.
#[pyclass]
#[gen_stub_pyclass]
pub struct ReceiveMessage {
diff --git a/foreign/python/src/send_message.rs
b/foreign/python/src/send_message.rs
index e63b1a2f7..bfa3cdd67 100644
--- a/foreign/python/src/send_message.rs
+++ b/foreign/python/src/send_message.rs
@@ -30,8 +30,6 @@ use pyo3_stub_gen::{
use crate::user_headers::py_user_headers_to_rust;
/// 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.
#[pyclass(from_py_object)]
#[gen_stub_pyclass]
pub struct SendMessage {
diff --git a/foreign/python/src/topic.rs b/foreign/python/src/topic.rs
index 178f90dd5..40a7db228 100644
--- a/foreign/python/src/topic.rs
+++ b/foreign/python/src/topic.rs
@@ -15,8 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::time::Duration;
-
use iggy::prelude::{
IggyByteSize, IggyExpiry as RustIggyExpiry, MaxTopicSize as
RustMaxTopicSize,
Partition as RustPartition, Topic as RustTopic, TopicDetails as
RustTopicDetails,
@@ -26,7 +24,7 @@ use pyo3::prelude::*;
use pyo3::types::PyDelta;
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum,
gen_stub_pymethods};
-use crate::consumer::py_delta_to_iggy_duration;
+use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration};
/// The expiry of the messages in a topic.
#[gen_stub_pyclass_complex_enum]
@@ -55,7 +53,14 @@ impl TryFrom<RustIggyExpiry> for IggyExpiry {
Ok(match expiry {
RustIggyExpiry::ServerDefault => IggyExpiry::ServerDefault(),
RustIggyExpiry::ExpireDuration(duration) =>
IggyExpiry::ExpireDuration {
- duration: iggy_duration_to_py_delta(duration.get_duration())?,
+ duration: Python::attach(|py| {
+ iggy_duration_to_py_delta(py, duration).map(|delta|
delta.unbind())
+ })
+ .map_err(|err| {
+ PyValueError::new_err(format!(
+ "topic message expiry duration does not fit within
timedelta bounds: {err}"
+ ))
+ })?,
},
RustIggyExpiry::NeverExpire => IggyExpiry::NeverExpire(),
})
@@ -93,26 +98,6 @@ impl TryFrom<&IggyExpiry> for RustIggyExpiry {
}
}
-fn iggy_duration_to_py_delta(duration: Duration) -> PyResult<Py<PyDelta>> {
- let days = duration.as_secs() / 86_400;
- let secs_of_day = duration.as_secs() % 86_400;
- Python::attach(|py| {
- PyDelta::new(
- py,
- days as i32,
- secs_of_day as i32,
- duration.subsec_micros() as i32,
- true,
- )
- .map(|delta| delta.unbind())
- .map_err(|err| {
- PyValueError::new_err(format!(
- "topic message expiry duration does not fit within timedelta
bounds: {err}"
- ))
- })
- })
-}
-
/// The maximum size of a topic.
#[gen_stub_pyclass_complex_enum]
#[pyclass]
diff --git a/foreign/python/tests/conftest.py b/foreign/python/tests/conftest.py
index aa54ff50d..3ab97065f 100644
--- a/foreign/python/tests/conftest.py
+++ b/foreign/python/tests/conftest.py
@@ -131,5 +131,10 @@ def pytest_collection_modifyitems(items):
path.name for path in Path(__file__).parent.glob("test_*.py")
}
for item in items:
+ # Tests explicitly marked as unit need no server; auto-marking them
+ # integration too would make `-m "not integration"` unable to select
+ # them.
+ if item.get_closest_marker("unit"):
+ continue
if any(module in item.nodeid for module in integration_modules):
item.add_marker(pytest.mark.integration)
diff --git a/foreign/python/tests/test_client_config.py
b/foreign/python/tests/test_client_config.py
new file mode 100644
index 000000000..78df23459
--- /dev/null
+++ b/foreign/python/tests/test_client_config.py
@@ -0,0 +1,459 @@
+# 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.
+
+"""
+Tests for the TCP client configuration surface.
+
+`TcpConfig`, `TcpReconnectionConfig` and `AutoLogin` mirror the Rust SDK
+types, so most of these assert that a value set from Python survives to the
+getters and that unset fields fall back to the Rust defaults. The last class
+proves the point of the configuration: with `auto_login` set, credentials are
+replayed on connect and no manual `login_user()` is needed.
+"""
+
+import ast
+from collections.abc import Callable
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+ AutoCommit,
+ AutoCommitAfter,
+ AutoCommitWhen,
+ AutoLogin,
+ IggyClient,
+ IggyExpiry,
+ TcpConfig,
+ TcpReconnectionConfig,
+)
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
[email protected]
+class TestAutoLogin:
+ """Test the credentials carried into the client."""
+
+ def test_disabled_has_no_username(self):
+ """Test that the disabled variant carries no credentials."""
+ auto_login = AutoLogin.disabled()
+
+ assert auto_login.enabled is False
+ assert auto_login.username is None
+
+ def test_username_password_exposes_username_only(self):
+ """Test that the username is readable back but the password is not."""
+ auto_login = AutoLogin.username_password("iggy", "secret")
+
+ assert auto_login.enabled is True
+ assert auto_login.username == "iggy"
+ assert "secret" not in repr(auto_login)
+
+ def test_personal_access_token_hides_the_token(self):
+ """Test that a token login exposes neither a username nor the token."""
+ auto_login = AutoLogin.personal_access_token("secret-token")
+
+ assert auto_login.enabled is True
+ assert auto_login.username is None
+ assert "secret-token" not in repr(auto_login)
+
+
[email protected]
+class TestTcpReconnectionConfig:
+ """Test the reconnection policy."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured policy reconnects forever, one second
apart."""
+ reconnection = TcpReconnectionConfig()
+
+ assert reconnection.enabled is True
+ assert reconnection.max_retries is None
+ assert reconnection.interval == timedelta(seconds=1)
+ assert reconnection.reestablish_after == timedelta(seconds=5)
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ reconnection = TcpReconnectionConfig(
+ enabled=False,
+ max_retries=10,
+ interval=timedelta(milliseconds=250),
+ reestablish_after=timedelta(seconds=30),
+ )
+
+ assert reconnection.enabled is False
+ assert reconnection.max_retries == 10
+ assert reconnection.interval == timedelta(milliseconds=250)
+ assert reconnection.reestablish_after == timedelta(seconds=30)
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the adjacent flags cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ TcpReconnectionConfig(True)
+
+ @pytest.mark.parametrize(
+ "construct",
+ [
+ lambda duration: TcpReconnectionConfig(interval=duration),
+ lambda duration: TcpReconnectionConfig(reestablish_after=duration),
+ ],
+ ids=["interval", "reestablish_after"],
+ )
+ @pytest.mark.parametrize(
+ "negative",
+ [timedelta(microseconds=-1), timedelta(seconds=-1),
timedelta(days=-1)],
+ )
+ def test_negative_duration_is_rejected(
+ self,
+ construct: Callable[[timedelta], TcpReconnectionConfig],
+ negative: timedelta,
+ ):
+ """Test that a negative duration fails at construction, not at
connect."""
+ with pytest.raises(ValueError, match="negative"):
+ construct(negative)
+
+ @pytest.mark.parametrize("out_of_range", [-1, 2**32])
+ def test_out_of_range_max_retries_is_rejected(self, out_of_range: int):
+ """Test that a retry count outside the wire range names the argument.
+
+ The conversion pyo3 does on its own raises OverflowError, which is not
a
+ ValueError and so escapes the handler a caller wraps construction in.
+ """
+ with pytest.raises(ValueError, match="max_retries"):
+ TcpReconnectionConfig(max_retries=out_of_range)
+
+ def test_zero_reestablish_after_is_allowed(self):
+ """Test that a zero cooldown is legal and readable back."""
+ reconnection = TcpReconnectionConfig(reestablish_after=timedelta(0))
+
+ assert reconnection.reestablish_after == timedelta(0)
+
+ def test_zero_interval_is_allowed_with_bounded_retries(self):
+ """Test that a zero interval is legal as a bounded fast-retry
policy."""
+ reconnection = TcpReconnectionConfig(interval=timedelta(0),
max_retries=5)
+
+ assert reconnection.interval == timedelta(0)
+
+ def test_zero_interval_is_allowed_when_reconnection_is_disabled(self):
+ """Test that a zero interval is legal when nothing ever reads it."""
+ reconnection = TcpReconnectionConfig(enabled=False,
interval=timedelta(0))
+
+ assert reconnection.interval == timedelta(0)
+
+ def test_zero_interval_with_unlimited_retries_is_rejected(self):
+ """Test that the combination that reconnects in a continuous loop
fails."""
+ with pytest.raises(ValueError, match="zero"):
+ TcpReconnectionConfig(interval=timedelta(0))
+
+ def test_very_long_interval_round_trips(self):
+ """Test that an interval beyond 68 years survives the i32 boundary."""
+ reconnection = TcpReconnectionConfig(interval=timedelta(days=30_000))
+
+ assert reconnection.interval == timedelta(days=30_000)
+
+ def test_maximum_interval_round_trips(self):
+ """Test that the largest timedelta survives the day conversion."""
+ reconnection =
TcpReconnectionConfig(interval=timedelta(days=999_999_999))
+
+ assert reconnection.interval == timedelta(days=999_999_999)
+
+
[email protected]
+class TestTcpConfig:
+ """Test the transport configuration."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured transport matches the Rust SDK
defaults."""
+ config = TcpConfig()
+
+ assert config.server_address == "127.0.0.1:8090"
+ assert config.auto_login.enabled is False
+ assert config.reconnection.enabled is True
+ assert config.heartbeat_interval == timedelta(seconds=5)
+ assert config.tls_enabled is False
+ assert config.tls_domain == ""
+ assert config.tls_ca_file is None
+ assert config.tls_validate_certificate is True
+ assert config.nodelay is False
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ config = TcpConfig(
+ server_address="localhost:8090",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ reconnection=TcpReconnectionConfig(max_retries=3),
+ heartbeat_interval=timedelta(seconds=15),
+ tls_enabled=True,
+ tls_domain="localhost",
+ tls_ca_file="ca.pem",
+ tls_validate_certificate=False,
+ nodelay=True,
+ )
+
+ assert config.server_address == "localhost:8090"
+ assert config.auto_login.username == "iggy"
+ assert config.reconnection.max_retries == 3
+ assert config.heartbeat_interval == timedelta(seconds=15)
+ assert config.tls_enabled is True
+ assert config.tls_domain == "localhost"
+ assert config.tls_ca_file == "ca.pem"
+ assert config.tls_validate_certificate is False
+ assert config.nodelay is True
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the address cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ TcpConfig("127.0.0.1:8090")
+
+ def test_repr_hides_the_password(self):
+ """Test that the password does not leak through repr."""
+ config = TcpConfig(auto_login=AutoLogin.username_password("iggy",
"secret"))
+
+ assert "secret" not in repr(config)
+
+ def test_repr_shows_every_field_as_python(self):
+ """Test that repr covers the TLS fields and parses as Python.
+
+ The TLS fields are the ones a handshake is debugged with, and a repr is
+ only worth printing if it can be pasted back into a constructor.
+ """
+ config = TcpConfig(
+ heartbeat_interval=timedelta(seconds=15),
+ tls_enabled=True,
+ tls_domain="localhost",
+ tls_ca_file="ca.pem",
+ tls_validate_certificate=False,
+ nodelay=True,
+ )
+
+ printed = repr(config)
+
+ assert 'tls_domain="localhost"' in printed
+ assert 'tls_ca_file="ca.pem"' in printed
+ assert "tls_validate_certificate=False" in printed
+ assert "nodelay=True" in printed
+ assert "heartbeat_interval=datetime.timedelta(seconds=15)" in printed
+ ast.parse(printed)
+
+ @pytest.mark.parametrize(
+ "invalid_address",
+ ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000",
"::1:8090"],
+ )
+ def test_invalid_server_address_is_rejected(self, invalid_address: str):
+ """Test that a malformed address fails at construction, not at
connect."""
+ with pytest.raises(ValueError):
+ TcpConfig(server_address=invalid_address)
+
+ def test_negative_heartbeat_interval_is_rejected(self):
+ """Test that a negative heartbeat interval fails at construction."""
+ with pytest.raises(ValueError, match="negative"):
+ TcpConfig(heartbeat_interval=timedelta(seconds=-3))
+
+ def test_zero_heartbeat_interval_is_rejected(self):
+ """Test that a zero heartbeat interval fails at construction.
+
+ Nothing downstream reads zero as "disabled"; it heartbeats in a
+ continuous loop for as long as the client lives.
+ """
+ with pytest.raises(ValueError, match="zero"):
+ TcpConfig(heartbeat_interval=timedelta(0))
+
+
[email protected]
+class TestClientConstruction:
+ """Test what the client constructor accepts."""
+
+ def test_accepts_a_config(self):
+ """Test that a client can be built from a config object."""
+ assert IggyClient(TcpConfig(server_address="127.0.0.1:8090")) is not
None
+
+ def test_accepts_an_address(self):
+ """Test that the address form still works."""
+ assert IggyClient("127.0.0.1:8090") is not None
+
+ def test_accepts_nothing(self):
+ """Test that the default address is used when no argument is given."""
+ assert IggyClient() is not None
+
+ def test_rejects_an_invalid_address(self):
+ """Test that a malformed address is rejected."""
+ with pytest.raises(RuntimeError):
+ IggyClient("nonsense")
+
+ def test_negative_message_expiry_is_rejected(self):
+ """Test that the negative-duration rule reaches create_topic.
+
+ The check runs at the call, before any I/O.
+ """
+ client = IggyClient()
+
+ with pytest.raises(ValueError, match="negative"):
+ client.create_topic(
+ stream="stream",
+ name="topic",
+ partitions_count=1,
+
message_expiry=IggyExpiry.ExpireDuration(timedelta(seconds=-1)),
+ )
+
+ @pytest.mark.parametrize(
+ "interval_kwargs",
+ [
+ {"polling_retry_interval": timedelta(0)},
+ {"init_retries": 3, "init_retry_interval": timedelta(0)},
+ {"auto_commit": AutoCommit.Interval(timedelta(0))},
+ {
+ "auto_commit": AutoCommit.IntervalOrWhen(
+ timedelta(0), AutoCommitWhen.PollingMessages()
+ )
+ },
+ {
+ "auto_commit": AutoCommit.IntervalOrAfter(
+ timedelta(0), AutoCommitAfter.ConsumingEachMessage()
+ )
+ },
+ ],
+ ids=[
+ "polling_retry_interval",
+ "init_retry_interval",
+ "auto_commit_interval",
+ "auto_commit_interval_or_when",
+ "auto_commit_interval_or_after",
+ ],
+ )
+ def test_zero_consumer_interval_is_rejected(self, interval_kwargs: dict):
+ """Test that a zero consumer interval fails at the call.
+
+ Zero spins the retry loop, floods the server with offset stores, or
+ panics inside the runtime timer, and none of those name the argument
+ that caused it.
+ """
+ client = IggyClient()
+
+ with pytest.raises(ValueError, match="zero"):
+ client.consumer_group(
+ name="group",
+ stream="stream",
+ topic="topic",
+ **interval_kwargs,
+ )
+
+ def test_zero_poll_interval_is_allowed(self):
+ """Test that a zero poll interval passes validation.
+
+ Zero there means "do not wait before polling" and is short-circuited
+ before the sleep, unlike the retry intervals. Reaching the awaitable is
+ what proves it: building one without a running loop is the next
failure,
+ and a rejected value would have raised ValueError first.
+ """
+ client = IggyClient()
+
+ with pytest.raises(RuntimeError):
+ client.consumer_group(
+ name="group",
+ stream="stream",
+ topic="topic",
+ poll_interval=timedelta(0),
+ )
+
+
[email protected]
+class TestAutoLoginAgainstServer:
+ """Test that configured credentials are actually replayed on connect."""
+
+ @pytest.mark.asyncio
+ async def test_auto_login_authenticates_without_login_user(self,
unique_name):
+ """Test that a privileged call succeeds without a manual
login_user()."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(
+ TcpConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ )
+ )
+ await client.connect()
+ await wait_for_ping(client)
+
+ stream_name = unique_name()
+ await client.create_stream(stream_name)
+ assert await client.get_stream(stream_name) is not None
+
+ @pytest.mark.asyncio
+ async def test_without_auto_login_a_privileged_call_is_unauthenticated(
+ self, unique_name
+ ):
+ """Test that the same call fails when no credentials are configured."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(TcpConfig(server_address=f"{host}:{port}"))
+ await client.connect()
+ await wait_for_ping(client)
+
+ with pytest.raises(RuntimeError):
+ await client.create_stream(unique_name())
+
+ @pytest.mark.asyncio
+ async def test_config_and_connection_string_both_authenticate(self,
unique_name):
+ """Test that either form of configuring credentials logs the client in.
+
+ The reconnection policy is set on both sides to mirror the connection
+ string, but the client exposes no getter for it, so this asserts only
+ what is observable: both clients reach an authenticated session.
+ """
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ from_config = IggyClient(
+ TcpConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ reconnection=TcpReconnectionConfig(
+ max_retries=3, interval=timedelta(seconds=1)
+ ),
+ )
+ )
+ from_string = IggyClient.from_connection_string(
+ f"iggy+tcp://iggy:iggy@{host}:{port}"
+ "?reconnection_retries=3&reconnection_interval=1s"
+ )
+
+ stream_name = unique_name()
+ for client in (from_config, from_string):
+ await client.connect()
+ await wait_for_ping(client)
+ assert await client.get_stream(stream_name) is None
+
+ @pytest.mark.asyncio
+ async def test_wrong_auto_login_credentials_fail(self):
+ """Test that bad configured credentials surface as a connect
failure."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(
+ TcpConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy",
"invalid-password"),
+ reconnection=TcpReconnectionConfig(enabled=False),
+ )
+ )
+
+ with pytest.raises(RuntimeError):
+ await client.connect()
