diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 90f4f98..7303a35 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 0a0b081..9901c4c 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 2a0d205..c4fb415 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 218bdf8..9c47f5f 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 cd91875..8c46285 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 41bb97a..ad0cf5f 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 4f718fa..a657857 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 0000000..da6b4bb --- /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 91fc46a..57b4387 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 7bcd9f4..42ef291 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 ab1bc61..61c4b39 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 b5b8914..10065d4 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"]