[Service Bus] Enable pylint and mypy

eng/tox/allowed_pylint_failures.py

@@ -39,7 +39,6 @@
     "azure-eventgrid",
     "azure-graphrbac",
     "azure-loganalytics",
-    "azure-servicebus",
     "azure-servicefabric",
     "azure-template",
     "azure-keyvault",

eng/tox/mypy_hard_failure_packages.py

@@ -8,6 +8,7 @@
 MYPY_HARD_FAILURE_OPTED = [
   "azure-core",
   "azure-eventhub",
+  "azure-servicebus",
   "azure-ai-textanalytics",
   "azure-ai-formrecognizer"
 ]

sdk/servicebus/azure-servicebus/azure/__init__.py

@@ -1 +1 @@
-__path__ = __import__('pkgutil').extend_path(__path__, __name__)
+__path__ = __import__('pkgutil').extend_path(__path__, __name__)  # type: ignore

sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py

@@ -110,7 +110,7 @@ def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
         return _generate_sas_token(scopes[0], self.policy, self.key)
 
 
-class BaseHandler(object):  # pylint:disable=too-many-instance-attributes
+class BaseHandler(object):
     def __init__(
         self,
         fully_qualified_namespace,
@@ -132,22 +132,6 @@ def __init__(
         self._auth_uri = None
         self._properties = create_properties()
 
-    def __enter__(self):
-        self._open_with_retry()
-        return self
-
-    def __exit__(self, *args):
-        self.close()
-
-    def _handle_exception(self, exception):
-        error, error_need_close_handler, error_need_raise = _create_servicebus_exception(_LOGGER, exception, self)
-        if error_need_close_handler:
-            self._close_handler()
-        if error_need_raise:
-            raise error
-
-        return error
-
     @staticmethod
     def _from_connection_string(conn_str, **kwargs):
         # type: (str, Any) -> Dict[str, Any]
@@ -173,6 +157,24 @@ def _from_connection_string(conn_str, **kwargs):
         kwargs["credential"] = ServiceBusSharedKeyCredential(policy, key)
         return kwargs
 
+
+class BaseHandlerSync(BaseHandler):  # pylint:disable=too-many-instance-attributes
+    def __enter__(self):
+        self._open_with_retry()
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    def _handle_exception(self, exception):
+        error, error_need_close_handler, error_need_raise = _create_servicebus_exception(_LOGGER, exception, self)
+        if error_need_close_handler:
+            self._close_handler()
+        if error_need_raise:
+            raise error
+
+        return error
+
     def _backoff(
         self,
         retried_times,

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

@@ -8,8 +8,8 @@
 import uuid
 import requests
 try:
-    from urlparse import urlparse
-    from urllib import unquote_plus
+    from urlparse import urlparse  # type: ignore
+    from urllib import unquote_plus  # type: ignore
 except ImportError:
     from urllib.parse import urlparse
     from urllib.parse import unquote_plus

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

@@ -8,10 +8,10 @@
 import uuid
 import functools
 import logging
-from typing import Optional, List, Union, Generator
+from typing import Optional, List, Union, Iterable, TYPE_CHECKING
 
-import uamqp
-from uamqp import types, errors
+import uamqp.message
+from uamqp import types
 
 from .constants import (
     _BATCH_MESSAGE_OVERHEAD_COST,
@@ -46,6 +46,9 @@
     MessageSettleFailed,
     MessageContentTooLarge)
 from .utils import utc_from_timestamp, utc_now
+if TYPE_CHECKING:
+    from .._servicebus_receiver import ServiceBusReceiver
+    from .._servicebus_session_receiver import ServiceBusSessionReceiver
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -86,8 +89,6 @@ def __init__(self, body, **kwargs):
         self._annotations = {}
         self._app_properties = {}
 
-        self._expiry = None
-        self._receiver = None
         self.session_id = kwargs.get("session_id", None)
         if 'message' in kwargs:
             self.message = kwargs['message']
@@ -268,10 +269,10 @@ def scheduled_enqueue_time_utc(self, value):
 
     @property
     def body(self):
-        # type: () -> Union[bytes, Generator[bytes]]
+        # type: () -> Union[bytes, Iterable[bytes]]
         """The body of the Message.
 
-        :rtype: bytes or generator[bytes]
+        :rtype: bytes or Iterable[bytes]
         """
         return self.message.get_data()
 
@@ -317,7 +318,8 @@ def __len__(self):
     def _from_list(self, messages):
         for each in messages:
             if not isinstance(each, Message):
-                raise ValueError("Populating a message batch only supports iterables containing Message Objects.  Received instead: {}".format(each.__class__.__name__))
+                raise ValueError("Populating a message batch only supports iterables containing Message Objects.  "
+                                 "Received instead: {}".format(each.__class__.__name__))
             self.add(each)
 
     @property
@@ -430,27 +432,79 @@ def sequence_number(self):
         return None
 
 
-class ReceivedMessage(PeekMessage):
-    """
-    A Service Bus Message received from service side.
-
-    :ivar auto_renew_error: Error when AutoLockRenew is used and it fails to renew the message lock.
-    :vartype auto_renew_error: ~azure.servicebus.AutoLockRenewTimeout or ~azure.servicebus.AutoLockRenewFailed
-
-    .. admonition:: Example:
-
-        .. literalinclude:: ../samples/sync_samples/sample_code_servicebus.py
-            :start-after: [START receive_complex_message]
-            :end-before: [END receive_complex_message]
-            :language: python
-            :dedent: 4
-            :caption: Checking the properties on a received message.
-    """
+class _ReceivedMessageBase(PeekMessage):
     def __init__(self, message, mode=ReceiveSettleMode.PeekLock, **kwargs):
-        super(ReceivedMessage, self).__init__(message=message)
+        super(_ReceivedMessageBase, self).__init__(message=message)
         self._settled = (mode == ReceiveSettleMode.ReceiveAndDelete)
         self._is_deferred_message = kwargs.get("is_deferred_message", False)
         self.auto_renew_error = None
+        self._receiver = None  # type: ignore
+        self._expiry = None
+
+    @property
+    def settled(self):
+        # type: () -> bool
+        """Whether the message has been settled.
+
+        This will aways be `True` for a message received using ReceiveAndDelete mode,
+        otherwise it will be `False` until the message is completed or otherwise settled.
+
+        :rtype: bool
+        """
+        return self._settled
+
+    @property
+    def expired(self):
+        # type: () -> bool
+        """
+
+        :rtype: bool
+        """
+        try:
+            if self._receiver.session:  # pylint: disable=protected-access
+                raise TypeError("Session messages do not expire. Please use the Session expiry instead.")
+        except AttributeError: # Is not a session receiver
+            pass
+        if self.locked_until_utc and self.locked_until_utc <= utc_now():
+            return True
+        return False
+
+    @property
+    def locked_until_utc(self):
+        # type: () -> Optional[datetime.datetime]
+        """
+
+        :rtype: datetime.datetime
+        """
+        try:
+            if self.settled or self._receiver.session:  # pylint: disable=protected-access
+                return None
+        except AttributeError: # not settled, and isn't session receiver.
+            pass
+        if self._expiry:
+            return self._expiry
+        if self.message.annotations and _X_OPT_LOCKED_UNTIL in self.message.annotations:
+            expiry_in_seconds = self.message.annotations[_X_OPT_LOCKED_UNTIL]/1000
+            self._expiry = utc_from_timestamp(expiry_in_seconds)
+        return self._expiry
+
+    @property
+    def lock_token(self):
+        # type: () -> Optional[Union[uuid.UUID, str]]
+        """
+
+        :rtype:  ~uuid.UUID or str
+        """
+        if self.settled:
+            return None
+
+        if self.message.delivery_tag:
+            return uuid.UUID(bytes_le=self.message.delivery_tag)
+
+        delivery_annotations = self.message.delivery_annotations
+        if delivery_annotations:
+            return delivery_annotations.get(_X_OPT_LOCK_TOKEN)
+        return None
 
     def _check_live(self, action):
         # pylint: disable=no-member
@@ -469,27 +523,6 @@ def _check_live(self, action):
         except AttributeError:
             pass
 
-    def _settle_message(
-            self,
-            settle_operation,
-            dead_letter_details=None
-    ):
-        try:
-            if not self._is_deferred_message:
-                try:
-                    self._settle_via_receiver_link(settle_operation, dead_letter_details)()
-                    return
-                except RuntimeError as exception:
-                    _LOGGER.info(
-                        "Message settling: %r has encountered an exception (%r)."
-                        "Trying to settle through management link",
-                        settle_operation,
-                        exception
-                    )
-            self._settle_via_mgmt_link(settle_operation, dead_letter_details)()
-        except Exception as e:
-            raise MessageSettleFailed(settle_operation, e)
-
     def _settle_via_mgmt_link(self, settle_operation, dead_letter_details=None):
         # pylint: disable=protected-access
         if settle_operation == MESSAGE_COMPLETE:
@@ -519,7 +552,11 @@ def _settle_via_mgmt_link(self, settle_operation, dead_letter_details=None):
             )
         raise ValueError("Unsupported settle operation type: {}".format(settle_operation))
 
-    def _settle_via_receiver_link(self, settle_operation, dead_letter_details=None):
+    def _settle_via_receiver_link(self, settle_operation, dead_letter_details=None):  # pylint: disable=unused-argument
+        # dead_letter_detail is not used because of uamqp receiver link doesn't accept it while it
+        # should be accepted. Will revisit this later.
+        # uamqp management link accepts dead_letter_details. Refer to method _settle_via_mgmt_link
+        # TODO: to make dead_letter_details useful
         if settle_operation == MESSAGE_COMPLETE:
             return functools.partial(self.message.accept)
         if settle_operation == MESSAGE_ABANDON:
@@ -532,70 +569,47 @@ def _settle_via_receiver_link(self, settle_operation, dead_letter_details=None):
             return functools.partial(self.message.modify, True, True)
         raise ValueError("Unsupported settle operation type: {}".format(settle_operation))
 
-    @property
-    def settled(self):
-        # type: () -> bool
-        """Whether the message has been settled.
 
-        This will aways be `True` for a message received using ReceiveAndDelete mode,
-        otherwise it will be `False` until the message is completed or otherwise settled.
+class ReceivedMessage(_ReceivedMessageBase):
+    """
+    A Service Bus Message received from service side.
 
-        :rtype: bool
-        """
-        return self._settled
+    :ivar auto_renew_error: Error when AutoLockRenew is used and it fails to renew the message lock.
+    :vartype auto_renew_error: ~azure.servicebus.AutoLockRenewTimeout or ~azure.servicebus.AutoLockRenewFailed
 
-    @property
-    def expired(self):
-        # type: () -> bool
-        """
+    .. admonition:: Example:
 
-        :rtype: bool
-        """
-        try:
-            if self._receiver.session:  # pylint: disable=protected-access
-                raise TypeError("Session messages do not expire. Please use the Session expiry instead.")
-        except AttributeError: # Is not a session receiver
-            pass
-        if self.locked_until_utc and self.locked_until_utc <= utc_now():
-            return True
-        return False
+        .. literalinclude:: ../samples/sync_samples/sample_code_servicebus.py
+            :start-after: [START receive_complex_message]
+            :end-before: [END receive_complex_message]
+            :language: python
+            :dedent: 4
+            :caption: Checking the properties on a received message.
+    """
 
-    @property
-    def locked_until_utc(self):
-        # type: () -> Optional[datetime.datetime]
-        """
+    def __init__(self, message, mode=ReceiveSettleMode.PeekLock, **kwargs):
+        super(ReceivedMessage, self).__init__(message, mode=mode, **kwargs)
 
-        :rtype: datetime.datetime
-        """
+    def _settle_message(
+            self,
+            settle_operation,
+            dead_letter_details=None
+    ):
         try:
-            if self.settled or self._receiver.session:  # pylint: disable=protected-access
-                return None
-        except AttributeError: # not settled, and isn't session receiver.
-            pass
-        if self._expiry:
-            return self._expiry
-        if self.message.annotations and _X_OPT_LOCKED_UNTIL in self.message.annotations:
-            expiry_in_seconds = self.message.annotations[_X_OPT_LOCKED_UNTIL]/1000
-            self._expiry = utc_from_timestamp(expiry_in_seconds)
-        return self._expiry
-
-    @property
-    def lock_token(self):
-        # type: () -> Optional[Union[uuid.UUID, str]]
-        """
-
-        :rtype:  ~uuid.UUID or str
-        """
-        if self.settled:
-            return None
-
-        if self.message.delivery_tag:
-            return uuid.UUID(bytes_le=self.message.delivery_tag)
-
-        delivery_annotations = self.message.delivery_annotations
-        if delivery_annotations:
-            return delivery_annotations.get(_X_OPT_LOCK_TOKEN)
-        return None
+            if not self._is_deferred_message:
+                try:
+                    self._settle_via_receiver_link(settle_operation, dead_letter_details)()
+                    return
+                except RuntimeError as exception:
+                    _LOGGER.info(
+                        "Message settling: %r has encountered an exception (%r)."
+                        "Trying to settle through management link",
+                        settle_operation,
+                        exception
+                    )
+            self._settle_via_mgmt_link(settle_operation, dead_letter_details)()
+        except Exception as e:
+            raise MessageSettleFailed(settle_operation, e)
 
     def complete(self):
         # type: () -> None
@@ -700,5 +714,5 @@ def renew_lock(self):
         if not token:
             raise ValueError("Unable to renew lock - no lock token found.")
 
-        expiry = self._receiver._renew_locks(token)  # pylint: disable=protected-access
+        expiry = self._receiver._renew_locks(token)  # pylint: disable=protected-access no-member
         self._expiry = utc_from_timestamp(expiry[MGMT_RESPONSE_MESSAGE_EXPIRATION][0]/1000.0)