ServiceBus dict-representation acceptance and kwarg-update functionality

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

@@ -42,7 +42,13 @@
     transform_messages_to_sendable_if_needed,
     trace_message,
 )
-from ..exceptions import MessageSizeExceededError
+from ..exceptions import (
+    MessageAlreadySettled,
+    MessageLockLostError,
+    SessionLockLostError,
+    MessageSizeExceededError,
+    ServiceBusError)
+from .utils import utc_from_timestamp, utc_now, transform_messages_to_sendable_if_needed, create_messages_from_dicts_if_needed
 
 if TYPE_CHECKING:
     from ..aio._servicebus_receiver_async import (
@@ -537,7 +543,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) and not isinstance(each, dict):
                 raise TypeError(
                     "Only ServiceBusMessage or an iterable object containing ServiceBusMessage "
                     "objects are accepted. Received instead: {}".format(
@@ -577,11 +583,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)[0]
         message = transform_messages_to_sendable_if_needed(message)
         trace_message(
             message, parent_span
@@ -629,6 +638,7 @@ class ServiceBusReceivedMessage(ServiceBusMessage):
 
     def __init__(self, message, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, **kwargs):
         # type: (uamqp.message.Message, Union[ServiceBusReceiveMode, str], Any) -> None
+        message = create_messages_from_dicts_if_needed(message, ServiceBusMessage)
         super(ServiceBusReceivedMessage, self).__init__(None, message=message)  # type: ignore
         self._settled = receive_mode == ServiceBusReceiveMode.RECEIVE_AND_DELETE
         self._received_timestamp_utc = utc_now()

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

@@ -9,7 +9,7 @@
 import logging
 import functools
 import platform
-from typing import Optional, Dict, Tuple, Iterable, Type, TYPE_CHECKING, Union, Iterator
+from typing import Optional, Dict, List, Tuple, Iterable, Type, TYPE_CHECKING, Union, Iterator
 from contextlib import contextmanager
 from msrest.serialization import UTC
 
@@ -43,10 +43,19 @@
 )
 
 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[
+        Dict,
+        ServiceBusMessage,
+        List[Dict],
+        List[ServiceBusMessage],
+        ServiceBusMessageBatch
+    ]
 
 _log = logging.getLogger(__name__)
 
@@ -196,6 +205,25 @@ def transform_messages_to_sendable_if_needed(messages):
         except AttributeError:
             return messages
 
+def create_messages_from_dicts_if_needed(messages, message_type):
+    """
+    This method is used to convert dict representations
+    of messages and to a list of ServiceBusMessage objects or ServiceBusBatchMessage.
+    """
+    # type: (DictMessageType) -> Union[List[azure.servicebus.ServiceBusMessage], azure.servicebus.ServiceBusMessageBatch]
+    if isinstance(messages, list):
+        for index, message in enumerate(messages):
+            if isinstance(message, dict):
+                messages[index] = message_type(**message)
+
+    if isinstance(messages, dict):
+        temp_messages = message_type(**messages)
+        messages = [temp_messages]
+
+    if isinstance(messages, message_type):
+        messages = [messages]
+
+    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)
         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()
+        messages = 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.")

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)
         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()
+        messages = 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.")

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_dicts_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_dicts_if_needed(queue, QueueProperties)
         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_dicts_if_needed(topic, TopicProperties)
         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_dicts_if_needed(subscription, SubscriptionProperties)
         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_dicts_if_needed(rule, RuleProperties)
         to_update = rule._to_internal_entity()
 
         create_entity_body = CreateRuleBody(

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_dicts_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_dicts_if_needed(queue, QueueProperties)
         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_dicts_if_needed(topic, TopicProperties)
         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_dicts_if_needed(subscription, SubscriptionProperties)
         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_dicts_if_needed(rule, RuleProperties)
         to_update = rule._to_internal_entity()
 
         create_entity_body = CreateRuleBody(

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

@@ -3,10 +3,28 @@
 # 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, Dict, List, Union
 from xml.etree.ElementTree import ElementTree, SubElement, QName
 import isodate
 import six
+import logging
+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
+    ]
+    DictPropertiesReturnType = Union[
+        InternalQueueDescription,
+        InternalTopicDescription,
+        InternalSubscriptionDescription,
+        InternalRuleDescription
+    ]
 
 # Refer to the async version of this module under ..\aio\management\_utils.py for detailed explanation.
 
@@ -18,6 +36,7 @@
 from azure.servicebus.management import _constants as constants
 from ._handle_response_error import _handle_response_error
 
+_log = logging.getLogger(__name__)
 
 def extract_rule_data_template(feed_class, convert, feed_element):
     """Special version of function extrat_data_template for Rule.
@@ -307,3 +326,15 @@ def _validate_topic_subscription_and_rule_types(
                 type(topic_name), type(subscription_name), type(rule_name)
             )
         )
+
+def create_properties_from_dicts_if_needed(properties, sb_resource_type):
+    """
+    This method is used to create a properties object given the 
+    resource properties type and its corresponding dict representation.
+    """
+    # type: (dict, DictPropertiesType) -> (DictPropertiesReturnType)
+    if isinstance(properties, dict):
+        dict_to_props = sb_resource_type(**properties)
+        properties = dict_to_props
+
+    return properties

sdk/servicebus/azure-servicebus/tests/async_tests/mgmt_tests/test_mgmt_queues_async.py

@@ -446,3 +446,70 @@ async def test_async_mgmt_queue_get_runtime_properties_negative(self, servicebus
 
         with pytest.raises(ResourceNotFoundError):
             await mgmt_service.get_queue_runtime_properties("non_existing_queue")
+
+    @pytest.mark.liveTest
+    @CachedResourceGroupPreparer(name_prefix='servicebustest')
+    @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
+    async def test_mgmt_queue_async_update_dict_success(self, servicebus_namespace_connection_string, **kwargs):
+        mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
+        await clear_queues(mgmt_service)
+        queue_name = "fjruid"
+        queue_description = await mgmt_service.create_queue(queue_name)
+        queue_description_dict = dict(queue_description)
+        try:
+            # Try updating one setting.
+            queue_description_dict["lock_duration"] = datetime.timedelta(minutes=2)
+            await mgmt_service.update_queue(queue_description_dict)
+
+            queue_description = await mgmt_service.get_queue(queue_name)
+            assert queue_description.lock_duration == datetime.timedelta(minutes=2)
+
+            # Now try updating all settings.
+            queue_description_dict = dict(queue_description)
+            queue_description_dict["auto_delete_on_idle"] = datetime.timedelta(minutes=10)
+            queue_description_dict["dead_lettering_on_message_expiration"] = True
+            queue_description_dict["default_message_time_to_live"] = datetime.timedelta(minutes=11)
+            queue_description_dict["duplicate_detection_history_time_window"] = datetime.timedelta(minutes=12)
+            queue_description_dict["enable_batched_operations"] = True
+            queue_description_dict["enable_express"] = True
+            #queue_description_dict["enable_partitioning"] = True # Cannot be changed after creation
+            queue_description_dict["lock_duration"] = datetime.timedelta(seconds=13)
+            queue_description_dict["max_delivery_count"] = 14
+            queue_description_dict["max_size_in_megabytes"] = 3072
+            #queue_description_dict["requires_duplicate_detection"] = True # Read only
+            #queue_description_dict["requires_session"] = True # Cannot be changed after creation
+
+            await mgmt_service.update_queue(queue_description_dict)
+            queue_description = await mgmt_service.get_queue(queue_name)
+
+            assert queue_description.auto_delete_on_idle == datetime.timedelta(minutes=10)
+            assert queue_description.dead_lettering_on_message_expiration == True
+            assert queue_description.default_message_time_to_live == datetime.timedelta(minutes=11)
+            assert queue_description.duplicate_detection_history_time_window == datetime.timedelta(minutes=12)
+            assert queue_description.enable_batched_operations == True
+            assert queue_description.enable_express == True
+            #assert queue_description.enable_partitioning == True
+            assert queue_description.lock_duration == datetime.timedelta(seconds=13)
+            assert queue_description.max_delivery_count == 14
+            assert queue_description.max_size_in_megabytes == 3072
+            #assert queue_description.requires_duplicate_detection == True
+            #assert queue_description.requires_session == True
+        finally:
+            await mgmt_service.delete_queue(queue_name)
+        await mgmt_service.close()
+
+    @pytest.mark.liveTest
+    @CachedResourceGroupPreparer(name_prefix='servicebustest')
+    @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
+    async def test_mgmt_queue_async_update_dict_error(self, servicebus_namespace_connection_string, **kwargs):
+        mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
+        await clear_queues(mgmt_service)
+        queue_name = "fjruid"
+        queue_description = await mgmt_service.create_queue(queue_name)
+        # send in queue dict without non-name keyword args
+        queue_description_only_name = {"name": queue_name}
+        with pytest.raises(TypeError):
+            await mgmt_service.update_queue(queue_description_only_name)
+
+        await mgmt_service.delete_queue(queue_name)
+        await mgmt_service.close()