Addresses warnings surfaced during p5 doc and API-view generation. (Azure#13069)

* Addresses warnings surfaced during p5 doc and API-view generation.  Removes localization from URIs, adds tons of type hints, normalizes return types, begins propogating class docstring into init docstring per guidelines.
* Fix backticks that were breaking link formatting in markdown in README.
* Add initial link for forever-receive refdocs inside readme.
* Address mypy issues.
* pylint fixes
* Defensive ServiceBusErrors for internal state misalignment.
* Adjust readme streaming URL to point to proper location.
* make receiver a ReceivedMessage param via kwargs and an expressive error to deincentivise use, but still required.

Co-authored-by: Adam Ling (MSFT) <[email protected]>

Author: KieranBrantnerMagee

sdk/servicebus/azure-servicebus/README.md

@@ -158,9 +158,9 @@ with ServiceBusClient.from_connection_string(connstr) as client:
 
 ### Receive messages from a queue
 
-To receive from a queue, you can either perform an ad-hoc receive via "receiver.receive_messages()" or receive persistently through the receiver itself.
+To receive from a queue, you can either perform an ad-hoc receive via `receiver.receive_messages()` or receive persistently through the receiver itself.
 
-#### Receive messages from a queue through iterating over ServiceBusReceiver
+#### [Receive messages from a queue through iterating over ServiceBusReceiver][streaming_receive_reference]
 
 ```Python
 from azure.servicebus import ServiceBusClient
@@ -173,7 +173,7 @@ with ServiceBusClient.from_connection_string(connstr) as client:
     # max_wait_time specifies how long the receiver should wait with no incoming messages before stopping receipt.  
     # Default is None; to receive forever.
     with client.get_queue_receiver(queue_name, max_wait_time=30) as receiver:
-        for msg in receiver:  # ServiceBusReceiver instance is a generator
+        for msg in receiver:  # ServiceBusReceiver instance is a generator. This is equivilent to get_streaming_message_iter().
             print(str(msg))
             # If it is desired to halt receiving early, one can break out of the loop here safely.
 ```
@@ -183,7 +183,7 @@ with ServiceBusClient.from_connection_string(connstr) as client:
 > See [AutoLockRenewer](#autolockrenew) for a helper to perform this in the background automatically.
 > Lock duration is set in Azure on the queue or topic itself.
 
-#### [Receive messages from a queue through `ServiceBusReceiver.receive_messages()`][receive_reference]
+#### [Receive messages from a queue through ServiceBusReceiver.receive_messages()][receive_reference]
 
 > **NOTE:** `ServiceBusReceiver.receive_messages()` receives a single or constrained list of messages through an ad-hoc method call, as opposed to receiving perpetually from the generator. It always returns a list.
 
@@ -462,6 +462,7 @@ contact [[email protected]](mailto:[email protected]) with any additio
 [client_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html#azure.servicebus.ServiceBusClient
 [send_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html?highlight=send_messages#azure.servicebus.ServiceBusSender.send_messages
 [receive_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html?highlight=receive#azure.servicebus.ServiceBusReceiver.receive_messages
+[streaming_receive_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html?highlight=get_streaming_message_iter#azure.servicebus.ServiceBusReceiver.get_streaming_message_iter
 [session_receive_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html?highlight=receive#azure.servicebus.ServiceBusSessionReceiver.receive_messages
 [session_send_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html?highlight=session_id#azure.servicebus.Message.session_id
 [complete_reference]: https://azuresdkdocs.blob.core.windows.net/$web/python/azure-servicebus/latest/azure.servicebus.html?highlight=complete#azure.servicebus.ReceivedMessage.complete

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

@@ -53,6 +53,17 @@ class AutoLockRenew(object):
     """
 
     def __init__(self, executor=None, max_workers=None):
+        # type: (ThreadPoolExecutor, int) -> None
+        """Auto renew locks for messages and sessions using a background thread pool.
+
+        :param executor: A user-specified thread pool. This cannot be combined with
+         setting `max_workers`.
+        :type executor: ~concurrent.futures.ThreadPoolExecutor
+        :param max_workers: Specify the maximum workers in the thread pool. If not
+         specified the number used will be derived from the core count of the environment.
+         This cannot be combined with `executor`.
+        :type max_workers: int
+        """
         self._executor = executor or ThreadPoolExecutor(max_workers=max_workers)
         self._shutdown = threading.Event()
         self._sleep_time = 1
@@ -109,16 +120,18 @@ def _auto_lock_renew(self, renewable, starttime, timeout, on_lock_renew_failure=
                 on_lock_renew_failure(renewable, error)
 
     def register(self, renewable, timeout=300, on_lock_renew_failure=None):
+        # type: (Union[ReceivedMessage, ServiceBusSession], float, Optional[LockRenewFailureCallback]) -> None
         """Register a renewable entity for automatic lock renewal.
 
         :param renewable: A locked entity that needs to be renewed.
-        :type renewable: ~azure.servicebus.ReceivedMessage or
-         ~azure.servicebus.ServiceBusSession
-        :param float timeout: A time in seconds that the lock should be maintained for.
-         Default value is 300 (5 minutes).
-        :param Optional[LockRenewFailureCallback] on_lock_renew_failure:
-         A callback may be specified to be called when the lock is lost on the renewable that is being registered.
-         Default value is None (no callback).
+        :type renewable: Union[~azure.servicebus.ReceivedMessage, ~azure.servicebus.ServiceBusSession]
+        :param timeout: A time in seconds that the lock should be maintained for. Default value is 300 (5 minutes).
+        :type timeout: float
+        :param on_lock_renew_failure: A callback may be specified to be called when the lock is lost on the renewable
+         that is being registered. Default value is None (no callback).
+        :type on_lock_renew_failure: Optional[LockRenewFailureCallback]
+
+        :rtype: None
         """
         if self._shutdown.is_set():
             raise ServiceBusError("The AutoLockRenew has already been shutdown. Please create a new instance for"
@@ -131,6 +144,8 @@ def close(self, wait=True):
 
         :param wait: Whether to block until thread pool has shutdown. Default is `True`.
         :type wait: bool
+
+        :rtype: None
         """
         self._shutdown.set()
         self._executor.shutdown(wait=wait)

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

@@ -9,7 +9,7 @@
 import uuid
 import functools
 import logging
-from typing import Optional, List, Union, Iterable, TYPE_CHECKING, Callable
+from typing import Optional, List, Union, Iterable, TYPE_CHECKING, Callable, Any
 
 import uamqp.message
 
@@ -52,7 +52,8 @@
     MessageLockExpired,
     SessionLockExpired,
     MessageSettleFailed,
-    MessageContentTooLarge)
+    MessageContentTooLarge,
+    ServiceBusError)
 from .utils import utc_from_timestamp, utc_now, copy_messages_to_sendable_if_needed
 if TYPE_CHECKING:
     from .._servicebus_receiver import ServiceBusReceiver
@@ -65,7 +66,7 @@ class Message(object):  # pylint: disable=too-many-public-methods,too-many-insta
     """A Service Bus Message.
 
     :param body: The data to send in a single message.
-    :type body: str or bytes
+    :type body: Union[str, bytes]
 
     :keyword dict properties: The user defined properties on the message.
     :keyword str session_id: The session identifier of the message for a sessionful entity.
@@ -95,6 +96,7 @@ class Message(object):  # pylint: disable=too-many-public-methods,too-many-insta
     """
 
     def __init__(self, body, **kwargs):
+        # type: (Union[str, bytes], Any) -> None
         # Although we might normally thread through **kwargs this causes
         # problems as MessageProperties won't absorb spurious args.
         self._encoding = kwargs.pop("encoding", 'UTF-8')
@@ -491,7 +493,6 @@ class BatchMessage(object):
     :vartype message: ~uamqp.BatchMessage
 
     :param int max_size_in_bytes: The maximum size of bytes data that a BatchMessage object can hold.
-
     """
     def __init__(self, max_size_in_bytes=None):
         # type: (Optional[int]) -> None
@@ -570,11 +571,11 @@ class PeekMessage(Message):
     This message is still on the queue, and unlocked.
     A peeked message cannot be completed, abandoned, dead-lettered or deferred.
     It has no lock token or expiry.
-
     """
 
     def __init__(self, message):
-        super(PeekMessage, self).__init__(None, message=message)
+        # type: (uamqp.message.Message) -> None
+        super(PeekMessage, self).__init__(None, message=message) # type: ignore
 
     def _to_outgoing_message(self):
         # type: () -> Message
@@ -741,12 +742,17 @@ class ReceivedMessageBase(PeekMessage):
     """
 
     def __init__(self, message, mode=ReceiveSettleMode.PeekLock, **kwargs):
+        # type: (uamqp.message.Message, ReceiveSettleMode, Any) -> None
         super(ReceivedMessageBase, self).__init__(message=message)
         self._settled = (mode == ReceiveSettleMode.ReceiveAndDelete)
         self._received_timestamp_utc = utc_now()
         self._is_deferred_message = kwargs.get("is_deferred_message", False)
-        self.auto_renew_error = None
-        self._receiver = None  # type: ignore
+        self.auto_renew_error = None # type: Optional[Exception]
+        try:
+            self._receiver = kwargs.pop("receiver")  # type: Union[ServiceBusReceiver, ServiceBusSessionReceiver]
+        except KeyError:
+            raise TypeError("ReceivedMessage requires a receiver to be initialized.  This class should never be" + \
+            "initialized by a user; the Message class should be utilized instead.")
         self._expiry = None
 
     def _check_live(self, action):
@@ -769,6 +775,7 @@ def _check_live(self, action):
     def _settle_via_mgmt_link(self, settle_operation, dead_letter_reason=None, dead_letter_description=None):
         # type: (str, Optional[str], Optional[str]) -> Callable
         # pylint: disable=protected-access
+
         if settle_operation == MESSAGE_COMPLETE:
             return functools.partial(
                 self._receiver._settle_message,
@@ -822,13 +829,14 @@ def _settle_via_receiver_link(self, settle_operation, dead_letter_reason=None, d
     @property
     def _lock_expired(self):
         # type: () -> bool
+        # pylint: disable=protected-access
         """
         Whether the lock on the message has expired.
 
         :rtype: bool
         """
         try:
-            if self._receiver.session:  # pylint: disable=protected-access
+            if self._receiver.session:  # type: ignore
                 raise TypeError("Session messages do not expire. Please use the Session expiry instead.")
         except AttributeError: # Is not a session receiver
             pass
@@ -859,6 +867,7 @@ def lock_token(self):
     @property
     def locked_until_utc(self):
         # type: () -> Optional[datetime.datetime]
+        # pylint: disable=protected-access
         """
         The UTC datetime until which the message will be locked in the queue/subscription.
         When the lock expires, delivery count of hte message is incremented and the message
@@ -867,7 +876,7 @@ def locked_until_utc(self):
         :rtype: datetime.datetime
         """
         try:
-            if self._settled or self._receiver.session:  # pylint: disable=protected-access
+            if self._settled or self._receiver.session:  # type: ignore
                 return None
         except AttributeError:  # not settled, and isn't session receiver.
             pass
@@ -1021,6 +1030,7 @@ def defer(self):
 
     def renew_lock(self):
         # type: () -> None
+        # pylint: disable=protected-access,no-member
         """Renew the message lock.
 
         This will maintain the lock on the message to ensure it is not returned to the queue
@@ -1041,7 +1051,7 @@ def renew_lock(self):
         :raises: ~azure.servicebus.exceptions.MessageAlreadySettled is message has already been settled.
         """
         try:
-            if self._receiver.session:
+            if self._receiver.session: # type: ignore
                 raise TypeError("Session messages cannot be renewed. Please renew the Session lock instead.")
         except AttributeError:
             pass
@@ -1050,5 +1060,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,no-member
+        expiry = self._receiver._renew_locks(token)  # type: ignore
         self._expiry = utc_from_timestamp(expiry[MGMT_RESPONSE_MESSAGE_EXPIRATION][0]/1000.0)

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

@@ -62,14 +62,15 @@ def deferred_message_op(
         status_code,
         message,
         description,
+        receiver,
         mode=ReceiveSettleMode.PeekLock,
         message_type=ReceivedMessage
 ):
     if status_code == 200:
         parsed = []
         for m in message.get_data()[b'messages']:
             wrapped = uamqp.Message.decode_from_bytes(bytearray(m[b'message']))
-            parsed.append(message_type(wrapped, mode, is_deferred_message=True))
+            parsed.append(message_type(wrapped, mode, is_deferred_message=True, receiver=receiver))
         return parsed
     if status_code in [202, 204]:
         return []

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

@@ -51,8 +51,7 @@ def _populate_attributes(self, **kwargs):
         self._max_wait_time = kwargs.get("max_wait_time", None)
 
     def _build_message(self, received, message_type=ReceivedMessage):
-        message = message_type(message=received, mode=self._mode)
-        message._receiver = self  # pylint: disable=protected-access
+        message = message_type(message=received, mode=self._mode, receiver=self)
         self._last_received_sequenced_number = message.sequence_number
         return message
 

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_client.py

@@ -430,7 +430,7 @@ def get_subscription_deadletter_receiver(self, topic_name, subscription_name, **
         )
 
     def get_subscription_session_receiver(self, topic_name, subscription_name, session_id=None, **kwargs):
-        # type: (str, str, str, Any) -> ServiceBusReceiver
+        # type: (str, str, str, Any) -> ServiceBusSessionReceiver
         """Get ServiceBusReceiver for the specific subscription under the topic.
 
         :param str topic_name: The name of specific Service Bus Topic the client connects to.

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

@@ -277,7 +277,7 @@ def _settle_message(self, settlement, lock_tokens, dead_letter_details=None):
         )
 
     def _renew_locks(self, *lock_tokens):
-        # type: (*str) -> Any
+        # type: (str) -> Any
         message = {MGMT_REQUEST_LOCK_TOKENS: types.AMQPArray(lock_tokens)}
         return self._mgmt_request_response_with_retry(
             REQUEST_RESPONSE_RENEWLOCK_OPERATION,
@@ -286,15 +286,25 @@ def _renew_locks(self, *lock_tokens):
         )
 
     def get_streaming_message_iter(self, max_wait_time=None):
+        # type: (float) -> Iterator[ReceivedMessage]
         """Receive messages from an iterator indefinitely, or if a max_wait_time is specified, until
         such a timeout occurs.
 
-        :param float max_wait_time: Maximum time to wait in seconds for the next message to arrive.
+        :param max_wait_time: Maximum time to wait in seconds for the next message to arrive.
          If no messages arrive, and no timeout is specified, this call will not return
          until the connection is closed. If specified, and no messages arrive for the
          timeout period, the iterator will stop.
+        :type max_wait_time: float
+        :rtype: Iterator[ReceivedMessage]
 
-         :rtype Iterator[ReceivedMessage]
+        .. admonition:: Example:
+
+            .. literalinclude:: ../samples/sync_samples/sample_code_servicebus.py
+                :start-after: [START receive_forever]
+                :end-before: [END receive_forever]
+                :language: python
+                :dedent: 4
+                :caption: Receive indefinitely from an iterator in streaming fashion.
         """
         return self._iter_contextual_wrapper(max_wait_time)
 
@@ -308,6 +318,7 @@ def from_connection_string(
         """Create a ServiceBusReceiver from a connection string.
 
         :param conn_str: The connection string of a Service Bus.
+        :type conn_str: str
         :keyword str queue_name: The path of specific Service Bus Queue the client connects to.
         :keyword str topic_name: The path of specific Service Bus Topic which contains the Subscription
          the client connects to.
@@ -384,7 +395,8 @@ def receive_messages(self, max_batch_size=None, max_wait_time=None):
          If no messages arrive, and no timeout is specified, this call will not return
          until the connection is closed. If specified, an no messages arrive within the
          timeout period, an empty list will be returned.
-        :rtype: list[~azure.servicebus.ReceivedMessage]
+
+        :rtype: List[~azure.servicebus.ReceivedMessage]
 
         .. admonition:: Example:
 
@@ -411,9 +423,9 @@ def receive_deferred_messages(self, sequence_numbers):
         When receiving deferred messages from a partitioned entity, all of the supplied
         sequence numbers must be messages from the same partition.
 
-        :param list[int] sequence_numbers: A list of the sequence numbers of messages that have been
+        :param List[int] sequence_numbers: A list of the sequence numbers of messages that have been
          deferred.
-        :rtype: list[~azure.servicebus.ReceivedMessage]
+        :rtype: List[~azure.servicebus.ReceivedMessage]
 
         .. admonition:: Example:
 
@@ -440,14 +452,12 @@ def receive_deferred_messages(self, sequence_numbers):
 
         self._populate_message_properties(message)
 
-        handler = functools.partial(mgmt_handlers.deferred_message_op, mode=self._mode)
+        handler = functools.partial(mgmt_handlers.deferred_message_op, mode=self._mode, receiver=self)
         messages = self._mgmt_request_response_with_retry(
             REQUEST_RESPONSE_RECEIVE_BY_SEQUENCE_NUMBER,
             message,
             handler
         )
-        for m in messages:
-            m._receiver = self  # pylint: disable=protected-access
         return messages
 
     def peek_messages(self, message_count=1, sequence_number=None):
@@ -460,7 +470,8 @@ def peek_messages(self, message_count=1, sequence_number=None):
         :param int message_count: The maximum number of messages to try and peek. The default
          value is 1.
         :param int sequence_number: A message sequence number from which to start browsing messages.
-        :rtype: list[~azure.servicebus.PeekMessage]
+
+        :rtype: List[~azure.servicebus.PeekMessage]
 
         .. admonition:: Example:
 

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

@@ -198,10 +198,10 @@ def schedule_messages(self, messages, schedule_time_utc):
         """Send Message or multiple Messages to be enqueued at a specific time.
         Returns a list of the sequence numbers of the enqueued messages.
         :param messages: The message or list of messages to schedule.
-        :type messages: ~azure.servicebus.Message or list[~azure.servicebus.Message]
+        :type messages: Union[~azure.servicebus.Message, List[~azure.servicebus.Message]]
         :param schedule_time_utc: The utc date and time to enqueue the messages.
         :type schedule_time_utc: ~datetime.datetime
-        :rtype: list[int]
+        :rtype: List[int]
 
         .. admonition:: Example:
 
@@ -266,6 +266,7 @@ def from_connection_string(
         """Create a ServiceBusSender from a connection string.
 
         :param conn_str: The connection string of a Service Bus.
+        :type conn_str: str
         :keyword str queue_name: The path of specific Service Bus Queue the client connects to.
          Only one of queue_name or topic_name can be provided.
         :keyword str topic_name: The path of specific Service Bus Topic the client connects to.
@@ -280,7 +281,8 @@ def from_connection_string(
          keys: `'proxy_hostname'` (str value) and `'proxy_port'` (int value).
          Additionally the following keys may also be present: `'username', 'password'`.
         :keyword str user_agent: If specified, this will be added in front of the built-in user agent string.
-        :rtype: ~azure.servicebus.ServiceBusSenderClient
+
+        :rtype: ~azure.servicebus.ServiceBusSender
 
         .. admonition:: Example: