[ServiceBus] Make sub-client initializers internal (add '_' to from_conn_str for receiver/sender) (Azure#14502)

* Make sub-client initializers internal (add '_' to from_conn_str for receiver/sender)
Remove via_partition_key until we add transactions as a whole.
Rename properties to application_properties
Rename amqp_message to amqp_annotated_message for consistency

* Fix remaining unit tests (an amqp_annotated_message and internal conn str rename were missed)

* fix test_sb_client  from_connection_string naming

* PR fixes; remove debug prints from a test, and add a line to changelog for label vs subject

* lint fix, unneeded import

* Update sdk/servicebus/azure-servicebus/samples/sync_samples/sample_code_servicebus.py

* Apply suggestions from code review

Rename Message->ServiceBusMessage in documentation to align with concurrent changes in master.

Co-authored-by: Adam Ling (MSFT) <adam_ling@outlook.com>

* Unify Breaking Changes sections in changelog for b8
Rename AMQPMessage to AMQPAnnotatedMessage to match property name and other SDKs.

Co-authored-by: Adam Ling (MSFT) <adam_ling@outlook.com>

Author: KieranBrantnerMagee

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -48,6 +48,13 @@ now raise more concrete exception other than `MessageSettleFailed` and `ServiceB
   - Changed `ServiceBusReceivedMessage.renew_lock` to `ServiceBusReceiver.renew_message_lock`
 * `AutoLockRenewer.register` now takes `ServiceBusReceiver` as a positional parameter.
 * Removed `encoding` support from `ServiceBusMessage`.
+* `ServiceBusMessage.amqp_message` has been renamed to `ServiceBusMessage.amqp_annotated_message` for cross-sdk consistency.
+* All `name` parameters in `ServiceBusAdministrationClient` are now precisely specified ala `queue_name` or `rule_name`
+* `ServiceBusMessage.via_partition_key` is no longer exposed, this is pending a full implementation of transactions as it has no external use.  If needed, the underlying value can still be accessed in `ServiceBusMessage.amqp_annotated_message.annotations`.
+* `ServiceBusMessage.properties` has been renamed to `ServiceBusMessage.application_properties` for consistency with service verbiage.
+* Sub-client (`ServiceBusSender` and `ServiceBusReceiver`) `from_connection_string` initializers have been made internal until needed.  Clients should be initialized from root `ServiceBusClient`.
+* `ServiceBusMessage.label` has been renamed to `ServiceBusMessage.subject`.
+* `ServiceBusMessage.amqp_annotated_message` has had its type renamed from `AMQPMessage` to `AMQPAnnotatedMessage`
 
 **BugFixes**
 

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

@@ -22,15 +22,13 @@
     _X_OPT_SEQUENCE_NUMBER,
     _X_OPT_ENQUEUE_SEQUENCE_NUMBER,
     _X_OPT_PARTITION_KEY,
-    _X_OPT_VIA_PARTITION_KEY,
     _X_OPT_LOCKED_UNTIL,
     _X_OPT_LOCK_TOKEN,
     _X_OPT_SCHEDULED_ENQUEUE_TIME,
     _X_OPT_DEAD_LETTER_SOURCE,
     PROPERTIES_DEAD_LETTER_REASON,
     PROPERTIES_DEAD_LETTER_ERROR_DESCRIPTION,
     ANNOTATION_SYMBOL_PARTITION_KEY,
-    ANNOTATION_SYMBOL_VIA_PARTITION_KEY,
     ANNOTATION_SYMBOL_SCHEDULED_ENQUEUE_TIME,
     ANNOTATION_SYMBOL_KEY_MAP
 )
@@ -49,22 +47,21 @@ class ServiceBusMessage(object):  # pylint: disable=too-many-public-methods,too-
     :param body: The data to send in a single message.
     :type body: Union[str, bytes]
 
-    :keyword dict properties: The user defined properties on the message.
+    :keyword dict application_properties: The user defined properties on the message.
     :keyword str session_id: The session identifier of the message for a sessionful entity.
     :keyword str message_id: The id to identify the message.
     :keyword datetime.datetime scheduled_enqueue_time_utc: The utc scheduled enqueue time to the message.
     :keyword datetime.timedelta time_to_live: The life duration of a message.
     :keyword str content_type: The content type descriptor.
     :keyword str correlation_id: The correlation identifier.
-    :keyword str label: The application specific label.
+    :keyword str subject: The application specific subject, sometimes referred to as label.
     :keyword str partition_key: The partition key for sending a message to a partitioned entity.
-    :keyword str via_partition_key: The partition key for sending a message into an entity via a partitioned
-     transfer queue.
     :keyword str to: The `to` address used for auto_forward chaining scenarios.
     :keyword str reply_to: The address of an entity to send replies to.
     :keyword str reply_to_session_id: The session identifier augmenting the `reply_to` address.
 
-    :ivar AMQPMessage amqp_message: Advanced use only.  The internal AMQP message payload that is sent or received.
+    :ivar AMQPAnnotatedMessage amqp_annotated_message: Advanced use only.
+        The internal AMQP message payload that is sent or received.
 
     .. admonition:: Example:
 
@@ -92,21 +89,21 @@ def __init__(self, body, **kwargs):
             self._amqp_header = self.message.header
         else:
             self._build_message(body)
-            self.properties = kwargs.pop("properties", None)
+            self.application_properties = kwargs.pop("application_properties", None)
             self.session_id = kwargs.pop("session_id", None)
             self.message_id = kwargs.get("message_id", None)
             self.content_type = kwargs.pop("content_type", None)
             self.correlation_id = kwargs.pop("correlation_id", None)
             self.to = kwargs.pop("to", None)
             self.reply_to = kwargs.pop("reply_to", None)
             self.reply_to_session_id = kwargs.pop("reply_to_session_id", None)
-            self.label = kwargs.pop("label", None)
+            self.subject = kwargs.pop("subject", None)
             self.scheduled_enqueue_time_utc = kwargs.pop("scheduled_enqueue_time_utc", None)
             self.time_to_live = kwargs.pop("time_to_live", None)
             self.partition_key = kwargs.pop("partition_key", None)
-            self.via_partition_key = kwargs.pop("via_partition_key", None)
-        # If message is the full message, amqp_message is the "public facing interface" for what we expose.
-        self.amqp_message = AMQPMessage(self.message) # type: AMQPMessage
+
+        # If message is the full message, amqp_annotated_message is the "public facing interface" for what we expose.
+        self.amqp_annotated_message = AMQPAnnotatedMessage(self.message) # type: AMQPAnnotatedMessage
 
     def __str__(self):
         return str(self.message)
@@ -167,16 +164,16 @@ def session_id(self, value):
         self._amqp_properties.group_id = value
 
     @property
-    def properties(self):
+    def application_properties(self):
         # type: () -> dict
         """The user defined properties on the message.
 
         :rtype: dict
         """
         return self.message.application_properties
 
-    @properties.setter
-    def properties(self, value):
+    @application_properties.setter
+    def application_properties(self, value):
         # type: (dict) -> None
         self.message.application_properties = value
 
@@ -207,33 +204,6 @@ def partition_key(self, value):
         # type: (str) -> None
         self._set_message_annotations(_X_OPT_PARTITION_KEY, value)
 
-    @property
-    def via_partition_key(self):
-        # type: () -> Optional[str]
-        """ The partition key for sending a message into an entity via a partitioned transfer queue.
-
-        If a message is sent via a transfer queue in the scope of a transaction, this value selects the transfer
-        queue partition: This is functionally equivalent to `partition_key` and ensures that messages are kept
-        together and in order as they are transferred.
-
-        See Transfers and Send Via in
-        `https://docs.microsoft.com/azure/service-bus-messaging/service-bus-transactions#transfers-and-send-via`.
-
-        :rtype: str
-        """
-        via_p_key = None
-        try:
-            via_p_key = self.message.annotations.get(_X_OPT_VIA_PARTITION_KEY) or \
-                self.message.annotations.get(ANNOTATION_SYMBOL_VIA_PARTITION_KEY)
-            return via_p_key.decode('UTF-8')
-        except (AttributeError, UnicodeDecodeError):
-            return via_p_key
-
-    @via_partition_key.setter
-    def via_partition_key(self, value):
-        # type: (str) -> None
-        self._set_message_annotations(_X_OPT_VIA_PARTITION_KEY, value)
-
     @property
     def time_to_live(self):
         # type: () -> Optional[datetime.timedelta]
@@ -349,9 +319,9 @@ def correlation_id(self, val):
         self._amqp_properties.correlation_id = val
 
     @property
-    def label(self):
+    def subject(self):
         # type: () -> str
-        """The application specific label.
+        """The application specific subject, sometimes referred to as a label.
 
         This property enables the application to indicate the purpose of the message to the receiver in a standardized
         fashion, similar to an email subject line.
@@ -363,8 +333,8 @@ def label(self):
         except (AttributeError, UnicodeDecodeError):
             return self._amqp_properties.subject
 
-    @label.setter
-    def label(self, val):
+    @subject.setter
+    def subject(self, val):
         # type: (str) -> None
         self._amqp_properties.subject = val
 
@@ -624,17 +594,16 @@ def _to_outgoing_message(self):
             body=body,
             content_type=self.content_type,
             correlation_id=self.correlation_id,
-            label=self.label,
+            subject=self.subject,
             message_id=self.message_id,
             partition_key=self.partition_key,
-            properties=self.properties,
+            application_properties=self.application_properties,
             reply_to=self.reply_to,
             reply_to_session_id=self.reply_to_session_id,
             session_id=self.session_id,
             scheduled_enqueue_time_utc=self.scheduled_enqueue_time_utc,
             time_to_live=self.time_to_live,
-            to=self.to,
-            via_partition_key=self.via_partition_key
+            to=self.to
         )
 
     @property
@@ -798,7 +767,7 @@ def locked_until_utc(self):
         return self._expiry
 
 
-class AMQPMessage(object):
+class AMQPAnnotatedMessage(object):
     """
     The internal AMQP message that this ServiceBusMessage represents.  Is read-only.
     """

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_receiver.py

@@ -2,7 +2,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
-import datetime
 import time
 import logging
 import functools
@@ -48,6 +47,7 @@
 
 
 if TYPE_CHECKING:
+    import datetime
     from azure.core.credentials import TokenCredential
 
 _LOGGER = logging.getLogger(__name__)
@@ -152,6 +152,8 @@ def __iter__(self):
         return self._iter_contextual_wrapper()
 
     def _iter_contextual_wrapper(self, max_wait_time=None):
+        """The purpose of this wrapper is to allow both state restoration (for multiple concurrent iteration)
+        and per-iter argument passing that requires the former."""
         # pylint: disable=protected-access
         original_timeout = None
         while True:
@@ -406,7 +408,7 @@ def get_streaming_message_iter(self, max_wait_time=None):
         return self._iter_contextual_wrapper(max_wait_time)
 
     @classmethod
-    def from_connection_string(
+    def _from_connection_string(
         cls,
         conn_str,
         **kwargs

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_sender.py

@@ -27,8 +27,7 @@
     MGMT_REQUEST_MESSAGE,
     MGMT_REQUEST_MESSAGES,
     MGMT_REQUEST_MESSAGE_ID,
-    MGMT_REQUEST_PARTITION_KEY,
-    MGMT_REQUEST_VIA_PARTITION_KEY
+    MGMT_REQUEST_PARTITION_KEY
 )
 
 if TYPE_CHECKING:
@@ -76,8 +75,6 @@ def _build_schedule_request(cls, schedule_time_utc, *messages):
                 message_data[MGMT_REQUEST_SESSION_ID] = message.session_id
             if message.partition_key:
                 message_data[MGMT_REQUEST_PARTITION_KEY] = message.partition_key
-            if message.via_partition_key:
-                message_data[MGMT_REQUEST_VIA_PARTITION_KEY] = message.via_partition_key
             message_data[MGMT_REQUEST_MESSAGE] = bytearray(message.message.encode_message())
             request_body[MGMT_REQUEST_MESSAGES].append(message_data)
         return request_body
@@ -271,7 +268,7 @@ def cancel_scheduled_messages(self, sequence_numbers, **kwargs):
         )
 
     @classmethod
-    def from_connection_string(
+    def _from_connection_string(
         cls,
         conn_str,
         **kwargs

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py

@@ -407,7 +407,7 @@ def get_streaming_message_iter(
         return self._IterContextualWrapper(self, max_wait_time)
 
     @classmethod
-    def from_connection_string(
+    def _from_connection_string(
         cls,
         conn_str: str,
         **kwargs: Any

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_sender_async.py

@@ -217,7 +217,7 @@ async def cancel_scheduled_messages(self, sequence_numbers: Union[int, List[int]
         )
 
     @classmethod
-    def from_connection_string(
+    def _from_connection_string(
         cls,
         conn_str: str,
         **kwargs: Any

sdk/servicebus/azure-servicebus/migration_guide.md

@@ -84,6 +84,11 @@ semantics with the sender or receiver lifetime.
 |---|---|---|
 | `azure.servicebus.AutoLockRenew().shutdown()` | `azure.servicebus.AutoLockRenewer().close()` | [Close an auto-lock-renewer](https://github.com/Azure/azure-sdk-for-python/tree/master/sdk/servicebus/azure-servicebus/samples/sync_samples/auto_lock_renew.py) |
 
+### Working with Message properties
+| In v0.50 | Equivalent in v7 | Sample |
+|---|---|---|
+| `azure.servicebus.Message.user_properties` | `azure.servicebus.ServiceBusMessage.application_properties` | Some message properties have been renamed, e.g. accessing the application specific properties of a message. |
+
 
 ## Migration samples
 

sdk/servicebus/azure-servicebus/samples/async_samples/sample_code_servicebus_async.py

@@ -57,7 +57,7 @@ async def example_create_servicebus_sender_async():
     from azure.servicebus.aio import ServiceBusSender
     servicebus_connection_str = os.environ['SERVICE_BUS_CONNECTION_STR']
     queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
-    queue_sender = ServiceBusSender.from_connection_string(
+    queue_sender = ServiceBusSender._from_connection_string(
         conn_str=servicebus_connection_str,
         queue_name=queue_name
     )
@@ -111,7 +111,7 @@ async def example_create_servicebus_receiver_async():
     from azure.servicebus.aio import ServiceBusReceiver
     servicebus_connection_str = os.environ['SERVICE_BUS_CONNECTION_STR']
     queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
-    queue_receiver = ServiceBusReceiver.from_connection_string(
+    queue_receiver = ServiceBusReceiver._from_connection_string(
         conn_str=servicebus_connection_str,
         queue_name=queue_name
     )

sdk/servicebus/azure-servicebus/samples/sync_samples/sample_code_servicebus.py

@@ -53,7 +53,7 @@ def example_create_servicebus_sender_sync():
     from azure.servicebus import ServiceBusSender
     servicebus_connection_str = os.environ['SERVICE_BUS_CONNECTION_STR']
     queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
-    queue_sender = ServiceBusSender.from_connection_string(
+    queue_sender = ServiceBusSender._from_connection_string(
         conn_str=servicebus_connection_str,
         queue_name=queue_name
     )
@@ -107,7 +107,7 @@ def example_create_servicebus_receiver_sync():
     from azure.servicebus import ServiceBusReceiver
     servicebus_connection_str = os.environ['SERVICE_BUS_CONNECTION_STR']
     queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
-    queue_receiver = ServiceBusReceiver.from_connection_string(
+    queue_receiver = ServiceBusReceiver._from_connection_string(
         conn_str=servicebus_connection_str,
         queue_name=queue_name
     )
@@ -244,7 +244,7 @@ def example_send_and_receive_sync():
             print("Sequence number: {}".format(message.sequence_number))
             print("Enqueued Sequence numger: {}".format(message.enqueued_sequence_number))
             print("Partition Key: {}".format(message.partition_key))
-            print("Properties: {}".format(message.properties))
+            print("Application Properties: {}".format(message.application_properties))
             print("Delivery count: {}".format(message.delivery_count))
             print("Message ID: {}".format(message.message_id))
             print("Locked until: {}".format(message.locked_until_utc))

sdk/servicebus/azure-servicebus/tests/async_tests/test_queues_async.py

@@ -432,8 +432,8 @@ async def test_async_queue_by_servicebus_client_iter_messages_with_retrieve_defe
                     print_message(_logger, message)
                     assert message.dead_letter_reason == 'Testing reason'
                     assert message.dead_letter_error_description == 'Testing description'
-                    assert message.properties[b'DeadLetterReason'] == b'Testing reason'
-                    assert message.properties[b'DeadLetterErrorDescription'] == b'Testing description'
+                    assert message.application_properties[b'DeadLetterReason'] == b'Testing reason'
+                    assert message.application_properties[b'DeadLetterErrorDescription'] == b'Testing description'
                     await receiver.complete_message(message)
             assert count == 10
 
@@ -549,8 +549,8 @@ async def test_async_queue_by_servicebus_client_receive_batch_with_deadletter(se
                     count += 1
                     assert message.dead_letter_reason == 'Testing reason'
                     assert message.dead_letter_error_description == 'Testing description'
-                    assert message.properties[b'DeadLetterReason'] == b'Testing reason'
-                    assert message.properties[b'DeadLetterErrorDescription'] == b'Testing description'
+                    assert message.application_properties[b'DeadLetterReason'] == b'Testing reason'
+                    assert message.application_properties[b'DeadLetterErrorDescription'] == b'Testing description'
                 assert count == 10
 
     @pytest.mark.liveTest
@@ -591,8 +591,8 @@ async def test_async_queue_by_servicebus_client_receive_batch_with_retrieve_dead
                     print_message(_logger, message)
                     assert message.dead_letter_reason == 'Testing reason'
                     assert message.dead_letter_error_description == 'Testing description'
-                    assert message.properties[b'DeadLetterReason'] == b'Testing reason'
-                    assert message.properties[b'DeadLetterErrorDescription'] == b'Testing description'
+                    assert message.application_properties[b'DeadLetterReason'] == b'Testing reason'
+                    assert message.application_properties[b'DeadLetterErrorDescription'] == b'Testing description'
                     await receiver.complete_message(message)
                     count += 1
             assert count == 10
@@ -1265,7 +1265,7 @@ def message_content():
                 for i in range(20):
                     yield ServiceBusMessage(
                         body="ServiceBusMessage no. {}".format(i),
-                        label='1st'
+                        subject='1st'
                     )
 
             sender = sb_client.get_queue_sender(servicebus_queue.name)
@@ -1291,12 +1291,12 @@ def message_content():
                     receive_counter += 1
                     for message in messages:
                         print_message(_logger, message)
-                        if message.label == '1st':
+                        if message.subject == '1st':
                             message_1st_received_cnt += 1
                             await receiver.complete_message(message)
-                            message.label = '2nd'
+                            message.subject = '2nd'
                             await sender.send_messages(message)  # resending received message
-                        elif message.label == '2nd':
+                        elif message.subject == '2nd':
                             message_2nd_received_cnt += 1
                             await receiver.complete_message(message)