Add ServiceBusReceivedMessage.from_bytes() classmethod (fixes #43979) (#45142)

* Add ServiceBusReceivedMessage.from_bytes() classmethod (fixes #43979)

Add a from_bytes() factory method to ServiceBusReceivedMessage that
constructs an instance from raw AMQP payload bytes without requiring
the deprecated uamqp library. This mirrors the existing
EventData.from_bytes() pattern in azure-eventhub.

The method uses the internal pyamqp decode_payload() to parse the
binary AMQP message and wraps it in an AmqpAnnotatedMessage, making
all standard message properties (body, application_properties,
annotations, etc.) accessible.

Includes unit tests for data, value, and sequence body types.

* Document settled state behavior in from_bytes docstring

* Bump version to 7.15.0 for unreleased changelog entry

* Fix mypy.ini python_version: 3.7 -> 3.9 (mypy 1.18+ requires >= 3.9)

* Disable mypy check for azure-servicebus: opt-out via pyproject.toml

mypy was effectively a no-op for this package because python_version=3.7
in mypy.ini caused mypy 1.18+ to abort immediately. Rather than fixing
61 pre-existing type errors across 4 files (unrelated to this PR), opt
out of the mypy check using the standard [tool.azure-sdk-build] mechanism
that many other packages already use. Revert mypy.ini to its original
state.

* Revert mypy=false from pyproject.toml (pre-existing CI issue, not related to this PR)

* Fix mypy errors: ignore pre-existing type issues in transport modules and message init

---------

Co-authored-by: Eldert Grootenboer (from Dev Box) <egrootenboer@microsoft.com>

Author: EldertGrootenboer

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Release History
 
-## 7.14.4 (Unreleased)
+## 7.15.0 (Unreleased)
+
+### Features Added
+
+- Added `ServiceBusReceivedMessage.from_bytes()` classmethod to construct a `ServiceBusReceivedMessage` from raw AMQP payload bytes without requiring the deprecated `uamqp` library. ([#43979](https://github.com/Azure/azure-sdk-for-python/issues/43979))
 
 ### Bugs Fixed
 

sdk/servicebus/azure-servicebus/api.md

@@ -299,6 +299,9 @@ namespace azure.servicebus
 
         def __str__(self) -> str: ...
 
+        @classmethod
+        def from_bytes(cls, message: bytes) -> ServiceBusReceivedMessage: ...
+
 
     class azure.servicebus.ServiceBusReceiver(BaseHandler, ReceiverMixin): implements ContextManager , Iterator 
         property client_identifier: str    # Read-only

sdk/servicebus/azure-servicebus/api.metadata.yml

@@ -1,3 +1,3 @@
-apiMdSha256: 0e51b69124cb208c793c257af32f007479fb277b28cf6d9b2bcb4e50cb4a5af7
+apiMdSha256: c5a1ed2bd1209332a754127f2464a9bdfa38a0ceaffca0a2c055b6405704224a
 parserVersion: 0.3.28
 pythonVersion: 3.12.10

sdk/servicebus/azure-servicebus/azure/servicebus/_common/message.py

@@ -9,6 +9,7 @@
 import time
 import warnings
 import datetime
+import struct
 import uuid
 from typing import Optional, Dict, List, Union, Iterable, Any, Mapping, cast, TYPE_CHECKING
 
@@ -808,6 +809,54 @@ def __getstate__(self) -> Dict[str, Any]:
     def __setstate__(self, state: Dict[str, Any]) -> None:
         self.__dict__.update(state)
 
+    @classmethod
+    def from_bytes(cls, message: bytes) -> "ServiceBusReceivedMessage":
+        """Constructs a ServiceBusReceivedMessage from the raw bytes of an AMQP message payload.
+
+        The message payload should adhere to the Message Format specification
+        outlined in the AMQP v1.0 standard:
+        http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format
+
+        The returned message has no associated receiver. It is intended for reading
+        message content and metadata, and cannot be used to settle the message
+        (e.g., complete, abandon, defer, dead_letter) or renew its lock, because those
+        operations require a live receiver.
+
+        Broker metadata carried in the payload is preserved. When the encoded message
+        contains the corresponding sections, properties such as ``lock_token``,
+        ``locked_until_utc``, ``sequence_number``, and ``enqueued_time_utc`` are
+        populated; when a section is absent, the corresponding property returns ``None``.
+
+        Note that AMQP value and sequence body sections, as well as some string-typed
+        properties, are surfaced as ``bytes`` after decoding (for example, a value body
+        of ``{"key": "value"}`` decodes to ``{b"key": b"value"}``). This reflects the
+        raw AMQP wire encoding and is not converted to ``str``.
+
+        :param bytes message: The raw bytes representing the AMQP message payload.
+        :return: A ServiceBusReceivedMessage created from the provided message payload.
+        :rtype: ~azure.servicebus.ServiceBusReceivedMessage
+        :raises TypeError: If ``message`` is not a ``bytes`` object.
+        :raises ValueError: If ``message`` is empty, or is not a valid AMQP 1.0
+         message payload (decode failures are wrapped and chained).
+        """
+        from .._pyamqp._decode import decode_payload
+
+        if not isinstance(message, bytes):
+            raise TypeError("message must be bytes.")
+        if not message:
+            raise ValueError("message cannot be empty.")
+
+        try:
+            amqp_message = decode_payload(memoryview(message))
+        except (ValueError, KeyError, IndexError, TypeError, EOFError, struct.error) as exc:
+            raise ValueError("message is not a valid AMQP 1.0 message payload.") from exc
+
+        received_msg = cls(
+            message=amqp_message,
+            receiver=None,
+        )
+        return received_msg
+
     @property
     def _lock_expired(self) -> bool:
         """

sdk/servicebus/azure-servicebus/azure/servicebus/_version.py

@@ -3,4 +3,4 @@
 # Licensed under the MIT License.
 # ------------------------------------
 
-VERSION = "7.14.4"
+VERSION = "7.15.0"

sdk/servicebus/azure-servicebus/tests/test_message.py

@@ -443,6 +443,190 @@ def test_servicebus_message_time_to_live():
     assert message.time_to_live == timedelta(days=1)
 
 
+def test_servicebus_received_message_from_bytes():
+    """Test that from_bytes can decode a basic AMQP message payload."""
+    from azure.servicebus._pyamqp._encode import encode_payload as _encode_payload
+    from azure.servicebus._pyamqp.message import Message as PyamqpMessage, Header, Properties
+
+    # Construct a pyamqp Message with various sections
+    original = PyamqpMessage(
+        header=Header(durable=True, priority=4, ttl=30000, delivery_count=1),
+        properties=Properties(
+            message_id="test-message-id-123",
+            content_type=b"application/json",
+            correlation_id="corr-456",
+            subject=b"test-subject",
+            reply_to=b"reply-queue",
+            group_id=b"session-1",
+            reply_to_group_id=b"reply-session",
+        ),
+        message_annotations={
+            _X_OPT_PARTITION_KEY: b"pk-1",
+        },
+        application_properties={b"custom_prop": b"custom_value"},
+        data=[b"hello world"],
+    )
+
+    # Encode to bytes then decode via from_bytes.
+    # _encode_payload returns the bytearray buffer it was given; wrap in bytes()
+    # so the test exercises the public from_bytes(bytes) contract.
+    output = bytearray()
+    payload = bytes(_encode_payload(output, original))
+    received = ServiceBusReceivedMessage.from_bytes(payload)
+
+    # Validate message body
+    body = b"".join(received.body)
+    assert body == b"hello world"
+    assert received.body_type == AmqpMessageBodyType.DATA
+
+    # Validate properties
+    assert received.message_id == "test-message-id-123"
+    assert received.content_type == "application/json"
+    assert received.correlation_id == "corr-456"
+    assert received.subject == "test-subject"
+    assert received.reply_to == "reply-queue"
+    assert received.session_id == "session-1"
+    assert received.reply_to_session_id == "reply-session"
+
+    # Validate header fields
+    assert received.time_to_live == timedelta(milliseconds=30000)
+    assert received.delivery_count == 1
+
+    # Validate annotations
+    assert received.partition_key == "pk-1"
+
+    # Validate application properties
+    assert received.raw_amqp_message.application_properties == {b"custom_prop": b"custom_value"}
+
+    # This payload carries no delivery-annotations lock token, so lock_token is None.
+    assert received.lock_token is None
+
+
+def test_servicebus_received_message_from_bytes_minimal():
+    """Test from_bytes with a minimal AMQP message (data body only, no properties)."""
+    from azure.servicebus._pyamqp._encode import encode_payload as _encode_payload
+    from azure.servicebus._pyamqp.message import Message as PyamqpMessage
+
+    original = PyamqpMessage(data=[b"minimal payload"])
+    output = bytearray()
+    payload = bytes(_encode_payload(output, original))
+    received = ServiceBusReceivedMessage.from_bytes(payload)
+
+    body = b"".join(received.body)
+    assert body == b"minimal payload"
+    assert received.message_id is None
+    assert received.content_type is None
+    assert received.session_id is None
+    assert received.lock_token is None
+
+
+def test_servicebus_received_message_from_bytes_value_body():
+    """Test from_bytes with an AMQP value body message."""
+    from azure.servicebus._pyamqp._encode import encode_payload as _encode_payload
+    from azure.servicebus._pyamqp.message import Message as PyamqpMessage
+
+    original = PyamqpMessage(value={"key": "value"})
+    output = bytearray()
+    payload = bytes(_encode_payload(output, original))
+    received = ServiceBusReceivedMessage.from_bytes(payload)
+
+    assert received.body_type == AmqpMessageBodyType.VALUE
+    # AMQP encoding round-trips strings as bytes
+    assert received.body == {b"key": b"value"}
+
+
+def test_servicebus_received_message_from_bytes_sequence_body():
+    """Test from_bytes with an AMQP sequence body message."""
+    from azure.servicebus._pyamqp._encode import encode_payload as _encode_payload
+    from azure.servicebus._pyamqp.message import Message as PyamqpMessage
+
+    original = PyamqpMessage(sequence=[1, 2, 3])
+    output = bytearray()
+    payload = bytes(_encode_payload(output, original))
+    received = ServiceBusReceivedMessage.from_bytes(payload)
+
+    assert received.body_type == AmqpMessageBodyType.SEQUENCE
+    # Confirm the decoded sequence contents round-trip, not just the body type.
+    assert list(received.body) == [1, 2, 3]
+
+
+def test_servicebus_received_message_from_bytes_raises_for_none():
+    """Test from_bytes raises a clear TypeError for non-bytes input."""
+    with pytest.raises(TypeError, match="message must be bytes"):
+        ServiceBusReceivedMessage.from_bytes(None)  # type: ignore[arg-type]
+
+
+def test_servicebus_received_message_from_bytes_raises_for_empty_payload():
+    """Test from_bytes rejects empty payloads with a clear ValueError."""
+    with pytest.raises(ValueError, match="message cannot be empty"):
+        ServiceBusReceivedMessage.from_bytes(b"")
+
+
+def test_servicebus_received_message_from_bytes_raises_for_invalid_payload():
+    """Test from_bytes wraps decode failures from malformed bytes in a chained ValueError.
+
+    Garbage bytes make the AMQP decoder raise low-level errors (e.g. ``TypeError`` /
+    ``IndexError``); ``from_bytes`` must surface a clear ``ValueError`` instead of
+    leaking those, and chain the original cause for diagnosability.
+    """
+    with pytest.raises(ValueError, match="not a valid AMQP") as exc_info:
+        ServiceBusReceivedMessage.from_bytes(b"\x01\x02\x03\x04")
+    assert exc_info.value.__cause__ is not None
+
+
+def test_servicebus_received_message_from_bytes_raises_for_truncated_payload():
+    """Test from_bytes wraps a truncated payload in a chained ValueError.
+
+    A partial read from the Functions host or a corrupted persisted message can
+    truncate a valid payload mid-section. That lands inside the fixed-width struct
+    decoders (e.g. ``_decode_ulong_large``), where ``unpack`` on a short slice raises
+    ``struct.error`` -- which derives from ``Exception``, not from ``IndexError``, so
+    a slice past the buffer end returns a truncated result rather than raising.
+    ``from_bytes`` must still surface a clear chained ``ValueError`` instead of leaking
+    the low-level ``struct.error``.
+    """
+    from azure.servicebus._pyamqp._encode import encode_payload as _encode_payload
+    from azure.servicebus._pyamqp.message import Message as PyamqpMessage, Header, Properties
+
+    original = PyamqpMessage(
+        header=Header(durable=True, priority=4, ttl=30000, delivery_count=1),
+        properties=Properties(message_id="test-message-id-123", content_type=b"application/json"),
+        application_properties={b"custom_prop": b"custom_value"},
+        data=[b"hello world"],
+    )
+    output = bytearray()
+    payload = bytes(_encode_payload(output, original))
+
+    # Cut partway through a fixed-width field so unpack sees a short buffer.
+    truncated = payload[:14]
+    with pytest.raises(ValueError, match="not a valid AMQP") as exc_info:
+        ServiceBusReceivedMessage.from_bytes(truncated)
+    assert exc_info.value.__cause__ is not None
+
+
+def test_servicebus_received_message_from_bytes_preserves_broker_metadata():
+    """Test from_bytes preserves broker metadata carried in the payload.
+
+    The payload's delivery-annotations carry ``x-opt-lock-token`` and its
+    message-annotations carry ``x-opt-sequence-number`` / ``x-opt-enqueued-time`` /
+    ``x-opt-locked-until``. Those properties must therefore be populated rather than
+    ``None``. This is a regression guard against forcing ``RECEIVE_AND_DELETE`` on
+    reconstruction, which would mark the message settled and suppress ``lock_token``
+    and ``locked_until_utc``.
+    """
+    # Real AMQP payload captured from the Azure Functions host in issue #43979.
+    payload = b'\x00Sp\xc0\x0b\x05@@pH\x19\x08\x00@R\x01\x00Sq\xc1$\x02\xa3\x10x-opt-lock-token\x98\xfcS\xa1_\xddfIO\x82]\xee\x1b|4<\xfb\x00Sr\xc1U\x06\xa3\x13x-opt-enqueued-time\x83\x00\x00\x01\x8ev\xc7\xdb\xc8\xa3\x15x-opt-sequence-numberU\x0c\xa3\x12x-opt-locked-until\x83\x00\x00\x01\x8ev\xc8\xc67\x00Ss\xc0?\r\xa1 f00d2a33551440389d68e299d31adc7c@@@@@@@\x83\x00\x00\x01\x8e\xbe\xe0\xe3\xc8\x83\x00\x00\x01\x8ev\xc7\xdb\xc8@@@\x00Su\xa0\x05hello'
+
+    received = ServiceBusReceivedMessage.from_bytes(payload)
+
+    assert b"".join(received.body) == b"hello"
+    assert received.message_id == "f00d2a33551440389d68e299d31adc7c"
+    assert received.sequence_number == 12
+    assert received.enqueued_time_utc is not None
+    assert received.locked_until_utc is not None
+    assert str(received.lock_token) == "fc53a15f-dd66-494f-825d-ee1b7c343cfb"
+
+
 def _pyamqp_encoded_inner_message(body=b"data"):
     # Encode a minimal AMQP message payload, as the service wraps each
     # deferred/peeked message inside the management-link response.