From 1a9e821669397268f9e46ae2833af9484013d55e Mon Sep 17 00:00:00 2001 From: emmanuel Date: Mon, 6 Jul 2026 09:32:27 -0700 Subject: [PATCH] Wrap awscrt types out of public MQTT signatures (#101) Introduce a library-owned QoS IntEnum and MqttConnectionHandle alias in nwp500/mqtt/types.py, translating to/from awscrt.mqtt.QoS only at the connection-layer boundary. Replace awscrt.mqtt.QoS and awscrt.mqtt.Connection type references across NavienMqttClient, MqttConnection, MqttSubscriptionManager, MqttCommandQueue and QueuedCommand. Wrap otherwise uncaught AwsCrtError in MqttPublishError at the public publish() boundary so awscrt exceptions no longer leak. Export QoS from nwp500 and nwp500.mqtt. Update tests to use the library QoS enum. --- CHANGELOG.rst | 23 ++++++++++++++ src/nwp500/__init__.py | 2 ++ src/nwp500/mqtt/__init__.py | 3 ++ src/nwp500/mqtt/client.py | 20 +++++++----- src/nwp500/mqtt/command_queue.py | 7 ++--- src/nwp500/mqtt/connection.py | 18 +++++------ src/nwp500/mqtt/subscriptions.py | 14 +++++---- src/nwp500/mqtt/types.py | 50 ++++++++++++++++++++++++++++++ src/nwp500/mqtt/utils.py | 5 ++- tests/test_command_queue.py | 7 ++--- tests/test_mqtt_reliability.py | 8 ++--- tests/test_protocol_correctness.py | 16 +++++----- 12 files changed, 125 insertions(+), 48 deletions(-) create mode 100644 src/nwp500/mqtt/types.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 90f4f986..7303a35e 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -25,6 +25,29 @@ Changed is byte-for-byte unchanged (verified by golden capture and the existing CLI tests); energy output now renders a summary table plus a breakdown table entirely in Rich rather than printing plain text and a Rich table. +- **awscrt types wrapped out of public MQTT signatures** (`#101 + `_): the library no + longer exposes ``awscrt`` SDK types on its own public/semi-public API + surface. A library-owned ``nwp500.mqtt.QoS`` ``IntEnum`` (also exported + as ``nwp500.QoS``) now replaces ``awscrt.mqtt.QoS`` on the ``publish`` + and ``subscribe`` methods of ``NavienMqttClient``, ``MqttConnection`` + and ``MqttSubscriptionManager``, on ``MqttCommandQueue.enqueue`` and on + the ``QueuedCommand`` dataclass. ``awscrt.mqtt.Connection`` handles are + typed behind the ``MqttConnectionHandle`` alias, and translation to/from + ``awscrt`` happens only at the connection-layer boundary + (``nwp500/mqtt/types.py``). ``NavienMqttClient.publish`` now wraps + otherwise-uncaught ``AwsCrtError`` in ``MqttPublishError`` so ``awscrt`` + exceptions no longer leak out of the public boundary. + + .. code-block:: python + + # OLD + from awscrt import mqtt + await client.publish(topic, payload, qos=mqtt.QoS.AT_LEAST_ONCE) + + # NEW + from nwp500 import QoS + await client.publish(topic, payload, qos=QoS.AT_LEAST_ONCE) Fixed ----- diff --git a/src/nwp500/__init__.py b/src/nwp500/__init__.py index 0a0b0812..9901c4ca 100644 --- a/src/nwp500/__init__.py +++ b/src/nwp500/__init__.py @@ -110,6 +110,7 @@ MqttMetrics, NavienMqttClient, PeriodicRequestType, + QoS, ) from nwp500.mqtt_events import ( MqttClientEvents, @@ -221,6 +222,7 @@ "MqttMetrics", "ConnectionDropEvent", "ConnectionEvent", + "QoS", # Event Emitter "EventEmitter", "EventListener", diff --git a/src/nwp500/mqtt/__init__.py b/src/nwp500/mqtt/__init__.py index 2a0d2058..c4fb4157 100644 --- a/src/nwp500/mqtt/__init__.py +++ b/src/nwp500/mqtt/__init__.py @@ -9,6 +9,7 @@ - PeriodicRequestType: Enum for periodic request types - MqttDiagnosticsCollector: Metrics and diagnostics collector - MqttMetrics, ConnectionDropEvent, ConnectionEvent: Diagnostic types +- QoS: Library-owned MQTT Quality of Service enum """ from .client import NavienMqttClient @@ -18,6 +19,7 @@ MqttDiagnosticsCollector, MqttMetrics, ) +from .types import QoS from .utils import MqttConnectionConfig, PeriodicRequestType __all__ = [ @@ -28,4 +30,5 @@ "MqttMetrics", "ConnectionDropEvent", "ConnectionEvent", + "QoS", ] diff --git a/src/nwp500/mqtt/client.py b/src/nwp500/mqtt/client.py index 218bdf8a..9c47f5f0 100644 --- a/src/nwp500/mqtt/client.py +++ b/src/nwp500/mqtt/client.py @@ -16,7 +16,6 @@ from collections.abc import Callable, Sequence from typing import TYPE_CHECKING, Any, cast -from awscrt import mqtt from awscrt.exceptions import AwsCrtError from ..auth import NavienAuthClient @@ -41,6 +40,7 @@ from .periodic import MqttPeriodicRequestManager from .reconnection import MqttReconnectionHandler from .subscriptions import MqttSubscriptionManager +from .types import MqttConnectionHandle, QoS from .utils import ( MqttConnectionConfig, PeriodicRequestType, @@ -246,7 +246,7 @@ def __init__( self._diagnostics = MqttDiagnosticsCollector() # Connection state (simpler than checking _connection_manager) - self._connection: mqtt.Connection | None = None + self._connection: MqttConnectionHandle | None = None self._connected = False # Guards _active_reconnect / _deep_reconnect against re-entrancy. # While True, _on_connection_interrupted_internal will not forward @@ -294,7 +294,7 @@ def _schedule_coroutine(self, coro: Any) -> None: future.add_done_callback(_log_scheduled_coroutine_result) def _on_connection_interrupted_internal( - self, connection: mqtt.Connection, error: AwsCrtError, **kwargs: Any + self, connection: MqttConnectionHandle, error: Exception, **kwargs: Any ) -> None: """Internal handler for connection interruption. @@ -350,7 +350,7 @@ def _on_connection_interrupted_internal( def _on_connection_resumed_internal( self, - connection: mqtt.Connection, + connection: MqttConnectionHandle, return_code: Any, session_present: Any, **kwargs: Any, @@ -885,7 +885,7 @@ async def subscribe( self, topic: str, callback: Callable[[str, dict[str, Any]], None], - qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE, + qos: QoS = QoS.AT_LEAST_ONCE, ) -> int: """ Subscribe to an MQTT topic. @@ -930,7 +930,7 @@ async def publish( self, topic: str, payload: dict[str, Any], - qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE, + qos: QoS = QoS.AT_LEAST_ONCE, ) -> int: """ Publish a message to an MQTT topic. @@ -990,9 +990,13 @@ async def publish( retriable=True, ) from e - # Other AWS CRT errors + # Other AWS CRT errors: wrap so awscrt exceptions never leak out + # of the public publish() boundary. _logger.error(f"Failed to publish to topic: {e}") - raise + raise MqttPublishError( + f"Failed to publish to MQTT topic: {e}", + retriable=True, + ) from e # Navien-specific convenience methods diff --git a/src/nwp500/mqtt/command_queue.py b/src/nwp500/mqtt/command_queue.py index cd918759..8c46285e 100644 --- a/src/nwp500/mqtt/command_queue.py +++ b/src/nwp500/mqtt/command_queue.py @@ -11,8 +11,7 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Any -from awscrt import mqtt - +from .types import QoS from .utils import QueuedCommand, redact_topic if TYPE_CHECKING: @@ -54,9 +53,7 @@ def __init__(self, config: MqttConnectionConfig): maxlen=config.max_queued_commands ) - def enqueue( - self, topic: str, payload: dict[str, Any], qos: mqtt.QoS - ) -> None: + def enqueue(self, topic: str, payload: dict[str, Any], qos: QoS) -> None: """ Add a command to the queue. diff --git a/src/nwp500/mqtt/connection.py b/src/nwp500/mqtt/connection.py index 41bb97a0..ad0cf5fe 100644 --- a/src/nwp500/mqtt/connection.py +++ b/src/nwp500/mqtt/connection.py @@ -14,7 +14,6 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any -from awscrt import mqtt from awscrt.exceptions import AwsCrtError from awsiot import mqtt_connection_builder @@ -22,6 +21,7 @@ MqttCredentialsError, MqttNotConnectedError, ) +from .types import MqttConnectionHandle, QoS, to_awscrt_qos if TYPE_CHECKING: from ..auth import NavienAuthClient @@ -50,10 +50,10 @@ def __init__( config: MqttConnectionConfig, auth_client: NavienAuthClient, on_connection_interrupted: ( - Callable[[mqtt.Connection, AwsCrtError], None] | None + Callable[[MqttConnectionHandle, Exception], None] | None ) = None, on_connection_resumed: ( - Callable[[mqtt.Connection, Any, Any | None], None] | None + Callable[[MqttConnectionHandle, Any, Any | None], None] | None ) = None, ): """ @@ -87,7 +87,7 @@ def __init__( self.config = config self._auth_client = auth_client - self._connection: mqtt.Connection | None = None + self._connection: MqttConnectionHandle | None = None self._connected = False self._on_connection_interrupted = on_connection_interrupted self._on_connection_resumed = on_connection_resumed @@ -350,7 +350,7 @@ async def close(self) -> None: async def subscribe( self, topic: str, - qos: mqtt.QoS, + qos: QoS, callback: Callable[..., None] | None = None, ) -> tuple[Any, int]: """ @@ -376,7 +376,7 @@ async def subscribe( # Use shield to prevent cancellation from propagating to # underlying future subscribe_future_raw, packet_id_raw = self._connection.subscribe( - topic=topic, qos=qos, callback=callback + topic=topic, qos=to_awscrt_qos(qos), callback=callback ) subscribe_future = subscribe_future_raw packet_id = packet_id_raw @@ -422,7 +422,7 @@ async def publish( self, topic: str, payload: str | dict[str, Any], - qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE, + qos: QoS = QoS.AT_LEAST_ONCE, ) -> int: """ Publish a message to an MQTT topic. @@ -453,7 +453,7 @@ async def publish( # Publish and get the concurrent.futures.Future publish_future_raw, packet_id_raw = self._connection.publish( - topic=topic, payload=payload_bytes, qos=qos + topic=topic, payload=payload_bytes, qos=to_awscrt_qos(qos) ) publish_future = publish_future_raw packet_id = int(packet_id_raw) @@ -488,7 +488,7 @@ def is_connected(self) -> bool: return self._connected @property - def connection(self) -> mqtt.Connection | None: + def connection(self) -> MqttConnectionHandle | None: """Get the underlying MQTT connection. Returns: diff --git a/src/nwp500/mqtt/subscriptions.py b/src/nwp500/mqtt/subscriptions.py index 4f718faa..a6578572 100644 --- a/src/nwp500/mqtt/subscriptions.py +++ b/src/nwp500/mqtt/subscriptions.py @@ -16,7 +16,6 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any, cast -from awscrt import mqtt from awscrt.exceptions import AwsCrtError from pydantic import ValidationError @@ -35,6 +34,7 @@ from ..mqtt_events import FeatureReceivedEvent, StatusReceivedEvent from ..topic_builder import MqttTopicBuilder from .state_tracker import DeviceStateTracker +from .types import QoS, to_awscrt_qos from .utils import get_response_data, redact_topic, topic_matches_pattern if TYPE_CHECKING: @@ -87,7 +87,7 @@ def __init__( self._operation_timeout = operation_timeout # Track subscriptions and handlers - self._subscriptions: dict[str, mqtt.QoS] = {} + self._subscriptions: dict[str, QoS] = {} self._message_handlers: dict[ str, list[Callable[[str, dict[str, Any]], None]] ] = {} @@ -96,7 +96,7 @@ def __init__( self._state_tracker = DeviceStateTracker(event_emitter) @property - def subscriptions(self) -> dict[str, mqtt.QoS]: + def subscriptions(self) -> dict[str, QoS]: """Get current subscriptions.""" return self._subscriptions.copy() @@ -205,7 +205,7 @@ async def subscribe( self, topic: str, callback: Callable[[str, dict[str, Any]], None], - qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE, + qos: QoS = QoS.AT_LEAST_ONCE, ) -> int: """ Subscribe to an MQTT topic. @@ -247,7 +247,9 @@ async def subscribe( # callbacks (avoids InvalidStateError); the underlying # subscribe completes independently. subscribe_future, packet_id = self._connection.subscribe( - topic=topic, qos=qos, callback=self._on_message_received + topic=topic, + qos=to_awscrt_qos(qos), + callback=self._on_message_received, ) awaitable_future = asyncio.wrap_future(subscribe_future) try: @@ -457,7 +459,7 @@ async def resubscribe_all(self) -> None: qos_map = dict(subscriptions_to_restore) for topic in failed_subscriptions: self._subscriptions[topic] = qos_map.get( - topic, mqtt.QoS.AT_LEAST_ONCE + topic, QoS.AT_LEAST_ONCE ) self._message_handlers[topic] = handlers_to_restore.get( topic, [] diff --git a/src/nwp500/mqtt/types.py b/src/nwp500/mqtt/types.py new file mode 100644 index 00000000..da6b4bb1 --- /dev/null +++ b/src/nwp500/mqtt/types.py @@ -0,0 +1,50 @@ +"""Library-owned MQTT transport types. + +This module defines transport-level types that the library exposes on its own +public and internal signatures instead of leaking ``awscrt`` SDK types. It is +the single boundary where the library's own :class:`QoS` enum is translated +to and from :class:`awscrt.mqtt.QoS`, keeping ``awscrt`` an implementation +detail that could be swapped for a different MQTT transport in the future. +""" + +from enum import IntEnum + +from awscrt import mqtt + +__author__ = "Emmanuel Levijarvi" +__copyright__ = "Emmanuel Levijarvi" +__license__ = "MIT" + + +class QoS(IntEnum): + """MQTT Quality of Service level. + + Mirrors the MQTT specification's QoS levels. Integer values match the + wire protocol (and :class:`awscrt.mqtt.QoS`) so translation is a direct + value mapping. + """ + + AT_MOST_ONCE = 0 + """QoS 0: fire-and-forget delivery with no acknowledgement.""" + + AT_LEAST_ONCE = 1 + """QoS 1: acknowledged delivery; duplicates possible.""" + + EXACTLY_ONCE = 2 + """QoS 2: acknowledged, exactly-once delivery.""" + + +# Opaque handle for the underlying MQTT connection object. Typed as an alias so +# callers and internal code refer to the library's name rather than the +# concrete ``awscrt`` type, keeping the transport swappable. +type MqttConnectionHandle = mqtt.Connection + + +def to_awscrt_qos(qos: QoS) -> mqtt.QoS: + """Translate a library :class:`QoS` to :class:`awscrt.mqtt.QoS`.""" + return mqtt.QoS(int(qos)) + + +def from_awscrt_qos(qos: mqtt.QoS) -> QoS: + """Translate an :class:`awscrt.mqtt.QoS` to a library :class:`QoS`.""" + return QoS(int(qos)) diff --git a/src/nwp500/mqtt/utils.py b/src/nwp500/mqtt/utils.py index 91fc46a0..57b43871 100644 --- a/src/nwp500/mqtt/utils.py +++ b/src/nwp500/mqtt/utils.py @@ -12,9 +12,8 @@ from enum import StrEnum from typing import Any -from awscrt import mqtt - from ..config import AWS_IOT_ENDPOINT, AWS_REGION +from .types import QoS __author__ = "Emmanuel Levijarvi" __copyright__ = "Emmanuel Levijarvi" @@ -286,7 +285,7 @@ class QueuedCommand: topic: str payload: dict[str, Any] - qos: mqtt.QoS + qos: QoS timestamp: datetime diff --git a/tests/test_command_queue.py b/tests/test_command_queue.py index 7bcd9f44..42ef291d 100644 --- a/tests/test_command_queue.py +++ b/tests/test_command_queue.py @@ -3,9 +3,8 @@ from collections import deque from datetime import UTC, datetime -from awscrt import mqtt - from nwp500.mqtt import MqttConnectionConfig +from nwp500.mqtt.types import QoS from nwp500.mqtt.utils import QueuedCommand @@ -13,7 +12,7 @@ def test_queued_command_dataclass(): """Test QueuedCommand dataclass creation.""" topic = "test/topic" payload = {"key": "value"} - qos = mqtt.QoS.AT_LEAST_ONCE + qos = QoS.AT_LEAST_ONCE timestamp = datetime.now(UTC) command = QueuedCommand( @@ -74,7 +73,7 @@ def test_queued_command_fifo_order(): command = QueuedCommand( topic=f"test/topic/{i}", payload={"value": i}, - qos=mqtt.QoS.AT_LEAST_ONCE, + qos=QoS.AT_LEAST_ONCE, timestamp=timestamp, ) queue.append(command) diff --git a/tests/test_mqtt_reliability.py b/tests/test_mqtt_reliability.py index ab1bc617..61c4b390 100644 --- a/tests/test_mqtt_reliability.py +++ b/tests/test_mqtt_reliability.py @@ -20,7 +20,6 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from awscrt import mqtt from nwp500.auth import AuthenticationResponse, AuthTokens, UserInfo from nwp500.exceptions import ( @@ -31,6 +30,7 @@ from nwp500.mqtt.connection import MqttConnection from nwp500.mqtt.periodic import MqttPeriodicRequestManager from nwp500.mqtt.reconnection import MqttReconnectionHandler +from nwp500.mqtt.types import QoS from nwp500.mqtt.utils import MqttConnectionConfig, PeriodicRequestType @@ -523,7 +523,7 @@ async def test_subscribe_times_out_without_ack(self, mock_auth_client): conn._connected = True with pytest.raises(TimeoutError): - await conn.subscribe("test/topic", qos=mqtt.QoS.AT_LEAST_ONCE) + await conn.subscribe("test/topic", qos=QoS.AT_LEAST_ONCE) @pytest.mark.asyncio async def test_publish_resolves_before_timeout(self, mock_auth_client): @@ -632,7 +632,7 @@ async def test_timeout_then_late_exception_is_consumed( conn._connected = True with pytest.raises(TimeoutError): - await conn.subscribe("test/topic", qos=mqtt.QoS.AT_LEAST_ONCE) + await conn.subscribe("test/topic", qos=QoS.AT_LEAST_ONCE) from awscrt.exceptions import AwsCrtError @@ -666,7 +666,7 @@ async def test_cancelled_then_late_exception_is_consumed( conn._connected = True task = asyncio.ensure_future( - conn.subscribe("test/topic", qos=mqtt.QoS.AT_LEAST_ONCE) + conn.subscribe("test/topic", qos=QoS.AT_LEAST_ONCE) ) await asyncio.sleep(0) task.cancel() diff --git a/tests/test_protocol_correctness.py b/tests/test_protocol_correctness.py index b5b89143..10065d48 100644 --- a/tests/test_protocol_correctness.py +++ b/tests/test_protocol_correctness.py @@ -14,7 +14,6 @@ from unittest.mock import AsyncMock, MagicMock import pytest -from awscrt import mqtt from nwp500.encoding import build_tou_period, encode_price from nwp500.events import EventEmitter @@ -28,6 +27,7 @@ from nwp500.mqtt.command_queue import MqttCommandQueue from nwp500.mqtt.control import MqttDeviceController from nwp500.mqtt.state_tracker import DeviceStateTracker +from nwp500.mqtt.types import QoS from nwp500.mqtt.utils import MqttConnectionConfig from nwp500.temperature import RawCelsius, TempFormulaType from nwp500.unit_system import reset_unit_system @@ -176,10 +176,8 @@ async def test_failed_command_requeued_at_front(self): newer commands, so [power_on, set_temp] became [set_temp, power_on] on the next flush.""" queue = _queue() - queue.enqueue( - "topic/power_on", {"cmd": "power"}, mqtt.QoS.AT_LEAST_ONCE - ) - queue.enqueue("topic/set_temp", {"cmd": "temp"}, mqtt.QoS.AT_LEAST_ONCE) + queue.enqueue("topic/power_on", {"cmd": "power"}, QoS.AT_LEAST_ONCE) + queue.enqueue("topic/set_temp", {"cmd": "temp"}, QoS.AT_LEAST_ONCE) publish = AsyncMock(side_effect=RuntimeError("still down")) @@ -207,8 +205,8 @@ class TestCommandQueueExpiry: @pytest.mark.asyncio async def test_expired_commands_discarded_on_flush(self): queue = _queue(max_age=300.0) - queue.enqueue("topic/old", {"cmd": "old"}, mqtt.QoS.AT_LEAST_ONCE) - queue.enqueue("topic/new", {"cmd": "new"}, mqtt.QoS.AT_LEAST_ONCE) + queue.enqueue("topic/old", {"cmd": "old"}, QoS.AT_LEAST_ONCE) + queue.enqueue("topic/new", {"cmd": "new"}, QoS.AT_LEAST_ONCE) # Age the first command past the limit queue._queue[0].timestamp = datetime.now(UTC) - timedelta(hours=2) @@ -222,7 +220,7 @@ async def test_expired_commands_discarded_on_flush(self): @pytest.mark.asyncio async def test_expiry_disabled_with_none(self): queue = _queue(max_age=None) - queue.enqueue("topic/old", {"cmd": "old"}, mqtt.QoS.AT_LEAST_ONCE) + queue.enqueue("topic/old", {"cmd": "old"}, QoS.AT_LEAST_ONCE) queue._queue[0].timestamp = datetime.now(UTC) - timedelta(days=1) publish = AsyncMock(return_value=1) @@ -236,7 +234,7 @@ def test_overflow_drops_oldest(self): ) queue = MqttCommandQueue(config) for name in ("a", "b", "c"): - queue.enqueue(f"topic/{name}", {}, mqtt.QoS.AT_LEAST_ONCE) + queue.enqueue(f"topic/{name}", {}, QoS.AT_LEAST_ONCE) assert queue.count == 2 assert [c.topic for c in queue._queue] == ["topic/b", "topic/c"]