diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index ee0dc06d64..51b34e34ab 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 bb10e91382..db0a6edb1f 100755 --- a/examples/python/getting-started/consumer.py +++ b/examples/python/getting-started/consumer.py @@ -19,8 +19,16 @@ 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 642399edb0..23f964fa81 100755 --- a/examples/python/getting-started/producer.py +++ b/examples/python/getting-started/producer.py @@ -19,8 +19,16 @@ 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 f709947b69..ff2cce834c 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 99a9b0ed68..8a5aff4250 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 341b3885c6..ab63e66c5a 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: ... +@typing.final +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`. """ +@typing.final +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: ... + +@typing.final +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 c7cbe39520..cde56040ff 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, + #[gen_stub(override_type(type_repr = "TcpConfig | builtins.str | None"))] conn: Option< + PyClientConfig, + >, ) -> PyResult { - 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::(e.to_string()) + })?, + ), + None => Arc::new(TcpClientConfig::default()), + }; + let tcp_client = TcpClient::create(config) .map_err(|e| PyErr::new::(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> { 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> { 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> { 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> { 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> { 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> { 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> { @@ -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::( @@ -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 0000000000..4519c939fb --- /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 { + 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, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option>, + ) -> PyResult { + 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 { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult> { + 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> { + 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, +} + +impl TcpConfig { + /// The configuration in the shape `TcpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc { + 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, + #[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>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + tls_validate_certificate: Option, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] nodelay: Option, + ) -> PyResult { + // 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> { + 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 { + 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 4d64fc626c..6a6e69a877 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 { 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) -> PyResult { + 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) -> PyResult { - 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 0000000000..9b2b419fe5 --- /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) -> PyResult { + 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::() + .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> { + 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 { + 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 985476ff74..9c7f1efaa7 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::()?; @@ -48,6 +51,9 @@ 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::()?; m.add_class::()?; m.add_class::()?; diff --git a/foreign/python/src/receive_message.rs b/foreign/python/src/receive_message.rs index ecabb6cf22..ddb14e6233 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 e63b1a2f78..bfa3cdd671 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 178f90dd5e..40a7db228e 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 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> { - 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 aa54ff50d8..3ab97065f5 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 0000000000..78df23459c --- /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 + + +@pytest.mark.unit +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) + + +@pytest.mark.unit +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) + + +@pytest.mark.unit +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)) + + +@pytest.mark.unit +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), + ) + + +@pytest.mark.integration +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()