ServiceBus dict-representation acceptance and kwarg-update functionality

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -2,6 +2,15 @@
 
 ## 7.0.2 (Unreleased)
 
+**New Features**
+
+* Updated the following so that they accept lists and single instances of dict representations for corresponding strongly-typed object arguments:
+  - `update_queue`, `update_topic`, `update_subscription`, and `update_rule` on `ServiceBusAdministrationClient` accept dict representations of `QueueProperties`, `TopicProperties`, `SubscriptionProperties`, and `RuleProperties`, respectively.
+  - `send_messages` and `schedule_messages` on both sync and async versions of `ServiceBusSender` accept a list of or single instance of dict representations of `ServiceBusMessage`.
+  - `add_message` on `ServiceBusMessageBatch` now accepts a dict representation of `ServiceBusMessage`.
+  - Note: This is ongoing work and is the first step in supporting the above as respresentation of type `typing.Mapping`.
+  - Note: Thanks to bradleydamato for their large contribution to this.
+
 **BugFixes**
 
 * Operations failing due to `uamqp.errors.LinkForceDetach` caused by no activity on the connection for 10 minutes will now be retried internally except for the session receiver case.

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

@@ -36,13 +36,14 @@
     MESSAGE_PROPERTY_MAX_LENGTH,
 )
 
+from ..exceptions import MessageSizeExceededError
 from .utils import (
     utc_from_timestamp,
     utc_now,
     transform_messages_to_sendable_if_needed,
     trace_message,
+    create_messages_from_dicts_if_needed
 )
-from ..exceptions import MessageSizeExceededError
 
 if TYPE_CHECKING:
     from ..aio._servicebus_receiver_async import (
@@ -537,7 +538,7 @@ def __len__(self):
     def _from_list(self, messages, parent_span=None):
         # type: (Iterable[ServiceBusMessage], AbstractSpan) -> None
         for each in messages:
-            if not isinstance(each, ServiceBusMessage):
+            if not isinstance(each, (ServiceBusMessage, dict)):
                 raise TypeError(
                     "Only ServiceBusMessage or an iterable object containing ServiceBusMessage "
                     "objects are accepted. Received instead: {}".format(
@@ -577,11 +578,14 @@ def add_message(self, message):
         :rtype: None
         :raises: :class: ~azure.servicebus.exceptions.MessageSizeExceededError, when exceeding the size limit.
         """
+
         return self._add(message)
 
     def _add(self, message, parent_span=None):
         # type: (ServiceBusMessage, AbstractSpan) -> None
         """Actual add implementation.  The shim exists to hide the internal parameters such as parent_span."""
+
+        message = create_messages_from_dicts_if_needed(message, ServiceBusMessage)  # type: ignore
         message = transform_messages_to_sendable_if_needed(message)
         trace_message(
             message, parent_span

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

@@ -9,7 +9,18 @@
 import logging
 import functools
 import platform
-from typing import Optional, Dict, Tuple, Iterable, Type, TYPE_CHECKING, Union, Iterator
+from typing import (
+    Any,
+    Dict,
+    Iterable,
+    Iterator,
+    List,
+    Mapping,
+    Optional,
+    Type,
+    TYPE_CHECKING,
+    Union
+)
 from contextlib import contextmanager
 from msrest.serialization import UTC
 
@@ -43,11 +54,26 @@
 )
 
 if TYPE_CHECKING:
-    from .message import ServiceBusReceivedMessage, ServiceBusMessage
+    from .message import ServiceBusReceivedMessage, ServiceBusMessage, ServiceBusMessageBatch
     from azure.core.tracing import AbstractSpan
     from .receiver_mixins import ReceiverMixin
     from .._servicebus_session import BaseSession
 
+    # pylint: disable=unused-import, ungrouped-imports
+    DictMessageType = Union[
+        Mapping,
+        ServiceBusMessage,
+        List[Mapping[str, Any]],
+        List[ServiceBusMessage],
+        ServiceBusMessageBatch
+    ]
+
+    DictMessageReturnType = Union[
+        ServiceBusMessage,
+        List[ServiceBusMessage],
+        ServiceBusMessageBatch
+    ]
+
 _log = logging.getLogger(__name__)
 
 
@@ -196,6 +222,20 @@ def transform_messages_to_sendable_if_needed(messages):
         except AttributeError:
             return messages
 
+def create_messages_from_dicts_if_needed(messages, message_type):
+    # type: (DictMessageType, type) -> DictMessageReturnType
+    """
+    This method is used to convert dict representations
+    of messages to a list of ServiceBusMessage objects or ServiceBusBatchMessage.
+    :param DictMessageType messages: A list or single instance of messages of type ServiceBusMessages or
+        dict representations of type ServiceBusMessage. Also accepts ServiceBusBatchMessage.
+    :rtype: DictMessageReturnType
+    """
+    if isinstance(messages, list):
+        return [(message_type(**message) if isinstance(message, dict) else message) for message in messages]
+
+    return_messages = message_type(**messages) if isinstance(messages, dict) else messages
+    return return_messages
 
 def strip_protocol_from_uri(uri):
     # type: (str) -> str

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

@@ -21,6 +21,7 @@
 from ._common.utils import (
     create_authentication,
     transform_messages_to_sendable_if_needed,
+    create_messages_from_dicts_if_needed,
     send_trace_context_manager,
     trace_message,
     add_link_to_send,
@@ -269,7 +270,9 @@ def schedule_messages(self, messages, schedule_time_utc, **kwargs):
                 :caption: Schedule a message to be sent in future
         """
         # pylint: disable=protected-access
+
         self._check_live()
+        messages = create_messages_from_dicts_if_needed(messages, ServiceBusMessage)    # type: ignore
         timeout = kwargs.pop("timeout", None)
         if timeout is not None and timeout <= 0:
             raise ValueError("The timeout must be greater than 0.")
@@ -365,7 +368,9 @@ def send_messages(self, message, **kwargs):
                 :caption: Send message.
 
         """
+
         self._check_live()
+        message = create_messages_from_dicts_if_needed(message, ServiceBusMessage)
         timeout = kwargs.pop("timeout", None)
         if timeout is not None and timeout <= 0:
             raise ValueError("The timeout must be greater than 0.")
@@ -393,11 +398,9 @@ def send_messages(self, message, **kwargs):
                 isinstance(message, ServiceBusMessageBatch) and len(message) == 0
             ):  # pylint: disable=len-as-condition
                 return  # Short circuit noop if an empty list or batch is provided.
-            if not isinstance(message, ServiceBusMessageBatch) and not isinstance(
-                message, ServiceBusMessage
-            ):
+            if not isinstance(message, (ServiceBusMessageBatch, ServiceBusMessage)):
                 raise TypeError(
-                    "Can only send azure.servicebus.<ServiceBusMessageBatch,ServiceBusMessage> "
+                    "Can only send azure.servicebus.<ServiceBusMessageBatch, ServiceBusMessage> "
                     "or lists of ServiceBusMessage."
                 )
 

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

@@ -22,6 +22,7 @@
 from .._common import mgmt_handlers
 from .._common.utils import (
     transform_messages_to_sendable_if_needed,
+    create_messages_from_dicts_if_needed,
     send_trace_context_manager,
     trace_message,
     add_link_to_send,
@@ -206,7 +207,9 @@ async def schedule_messages(
                 :caption: Schedule a message to be sent in future
         """
         # pylint: disable=protected-access
+
         self._check_live()
+        messages = create_messages_from_dicts_if_needed(messages, ServiceBusMessage)    # type: ignore
         timeout = kwargs.pop("timeout", None)
         if timeout is not None and timeout <= 0:
             raise ValueError("The timeout must be greater than 0.")
@@ -307,7 +310,9 @@ async def send_messages(
                 :caption: Send message.
 
         """
+
         self._check_live()
+        message = create_messages_from_dicts_if_needed(message, ServiceBusMessage)
         timeout = kwargs.pop("timeout", None)
         if timeout is not None and timeout <= 0:
             raise ValueError("The timeout must be greater than 0.")
@@ -333,9 +338,7 @@ async def send_messages(
                 isinstance(message, ServiceBusMessageBatch) and len(message) == 0
             ):  # pylint: disable=len-as-condition
                 return  # Short circuit noop if an empty list or batch is provided.
-            if not isinstance(message, ServiceBusMessageBatch) and not isinstance(
-                message, ServiceBusMessage
-            ):
+            if not isinstance(message, (ServiceBusMessageBatch, ServiceBusMessage)):
                 raise TypeError(
                     "Can only send azure.servicebus.<ServiceBusMessageBatch,ServiceBusMessage> "
                     "or lists of ServiceBusMessage."

sdk/servicebus/azure-servicebus/azure/servicebus/aio/management/_management_client_async.py

@@ -77,6 +77,7 @@
 from ...management._utils import (
     deserialize_rule_key_values,
     serialize_rule_key_values,
+    create_properties_from_dict_if_needed,
     _validate_entity_name_type,
     _validate_topic_and_subscription_types,
     _validate_topic_subscription_and_rule_types,
@@ -411,7 +412,9 @@ async def update_queue(self, queue: QueueProperties, **kwargs) -> None:
         :rtype: None
         """
 
+        queue = create_properties_from_dict_if_needed(queue, QueueProperties)   # type: ignore
         to_update = queue._to_internal_entity()
+
         to_update.default_message_time_to_live = avoid_timedelta_overflow(
             to_update.default_message_time_to_live
         )
@@ -636,6 +639,7 @@ async def update_topic(self, topic: TopicProperties, **kwargs) -> None:
         :rtype: None
         """
 
+        topic = create_properties_from_dict_if_needed(topic, TopicProperties)   # type: ignore
         to_update = topic._to_internal_entity()
 
         to_update.default_message_time_to_live = avoid_timedelta_overflow(
@@ -880,8 +884,10 @@ async def update_subscription(
          from `get_subscription`, `update_subscription` or `list_subscription` and has the updated properties.
         :rtype: None
         """
+
         _validate_entity_name_type(topic_name, display_name="topic_name")
 
+        subscription = create_properties_from_dict_if_needed(subscription, SubscriptionProperties)  # type: ignore
         to_update = subscription._to_internal_entity()
 
         to_update.default_message_time_to_live = avoid_timedelta_overflow(
@@ -1079,6 +1085,7 @@ async def update_rule(
         """
         _validate_topic_and_subscription_types(topic_name, subscription_name)
 
+        rule = create_properties_from_dict_if_needed(rule, RuleProperties)  # type: ignore
         to_update = rule._to_internal_entity()
 
         create_entity_body = CreateRuleBody(

sdk/servicebus/azure-servicebus/azure/servicebus/aio/management/_utils.py

@@ -8,7 +8,7 @@
 
 import urllib.parse as urlparse
 
-from azure.servicebus.management import _constants as constants
+from ...management import _constants as constants
 from ...management._handle_response_error import _handle_response_error
 
 # This module defines functions get_next_template and extract_data_template.

sdk/servicebus/azure-servicebus/azure/servicebus/management/_management_client.py

@@ -46,6 +46,7 @@
     deserialize_rule_key_values,
     serialize_rule_key_values,
     extract_rule_data_template,
+    create_properties_from_dict_if_needed,
     _validate_entity_name_type,
     _validate_topic_and_subscription_types,
     _validate_topic_subscription_and_rule_types,
@@ -404,6 +405,7 @@ def update_queue(self, queue, **kwargs):
         :rtype: None
         """
 
+        queue = create_properties_from_dict_if_needed(queue, QueueProperties)   # type: ignore
         to_update = queue._to_internal_entity()
 
         to_update.default_message_time_to_live = avoid_timedelta_overflow(
@@ -632,6 +634,7 @@ def update_topic(self, topic, **kwargs):
         :rtype: None
         """
 
+        topic = create_properties_from_dict_if_needed(topic, TopicProperties)   # type: ignore
         to_update = topic._to_internal_entity()
 
         to_update.default_message_time_to_live = (
@@ -884,8 +887,10 @@ def update_subscription(self, topic_name, subscription, **kwargs):
          from `get_subscription`, `update_subscription` or `list_subscription` and has the updated properties.
         :rtype: None
         """
+
         _validate_entity_name_type(topic_name, display_name="topic_name")
 
+        subscription = create_properties_from_dict_if_needed(subscription, SubscriptionProperties)  # type: ignore
         to_update = subscription._to_internal_entity()
 
         to_update.default_message_time_to_live = avoid_timedelta_overflow(
@@ -1077,6 +1082,7 @@ def update_rule(self, topic_name, subscription_name, rule, **kwargs):
         """
         _validate_topic_and_subscription_types(topic_name, subscription_name)
 
+        rule = create_properties_from_dict_if_needed(rule, RuleProperties)  # type: ignore
         to_update = rule._to_internal_entity()
 
         create_entity_body = CreateRuleBody(

sdk/servicebus/azure-servicebus/azure/servicebus/management/_utils.py

@@ -3,22 +3,39 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 from datetime import datetime, timedelta
-from typing import cast
+from typing import TYPE_CHECKING, cast, Union, Mapping
 from xml.etree.ElementTree import ElementTree, SubElement, QName
 import isodate
 import six
 
+from . import _constants as constants
+from ._handle_response_error import _handle_response_error
+if TYPE_CHECKING:
+    # pylint: disable=unused-import, ungrouped-imports
+    from ._models import QueueProperties, TopicProperties, \
+        SubscriptionProperties, RuleProperties, InternalQueueDescription, InternalTopicDescription, \
+        InternalSubscriptionDescription, InternalRuleDescription
+    DictPropertiesType = Union[
+        QueueProperties,
+        TopicProperties,
+        SubscriptionProperties,
+        RuleProperties,
+        Mapping
+    ]
+    DictPropertiesReturnType = Union[
+        QueueProperties,
+        TopicProperties,
+        SubscriptionProperties,
+        RuleProperties
+    ]
+
 # Refer to the async version of this module under ..\aio\management\_utils.py for detailed explanation.
 
 try:
     import urllib.parse as urlparse
 except ImportError:
     import urlparse  # type: ignore  # for python 2.7
 
-from azure.servicebus.management import _constants as constants
-from ._handle_response_error import _handle_response_error
-
-
 def extract_rule_data_template(feed_class, convert, feed_element):
     """Special version of function extrat_data_template for Rule.
 
@@ -307,3 +324,16 @@ def _validate_topic_subscription_and_rule_types(
                 type(topic_name), type(subscription_name), type(rule_name)
             )
         )
+
+def create_properties_from_dict_if_needed(properties, sb_resource_type):
+    # type: (DictPropertiesType, type) -> DictPropertiesReturnType
+    """
+    This method is used to create a properties object given the
+    resource properties type and its corresponding dict representation.
+    :param properties: A properties object or its dict representation.
+    :type properties: DictPropertiesType
+    :param type sb_resource_type: The type of properties object.
+    :rtype: DictPropertiesReturnType
+    """
+    return_properties = sb_resource_type(**properties) if isinstance(properties, dict) else properties
+    return return_properties