[Service Bus] Add batch delete and purge APIs

eng/ci_tools.txt

@@ -8,7 +8,8 @@ doc-warden==0.7.2
 beautifulsoup4==4.13.3
 pkginfo==1.12.1.2
 pip==26.2.2
-typing-extensions==4.12.2
+typing-extensions==4.12.2; python_version < "3.15"
+typing-extensions==4.14.0; python_version >= "3.15"
 pyproject-api==1.8.0
 cibuildwheel
 importlib-metadata==8.5.0

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Release History
 
+## 7.16.0b1 (Unreleased)
+
+### Features Added
+
+- Added sync and async `ServiceBusReceiver.delete_messages()` and `purge_messages()` methods. Basic and Standard support up to 500 messages per request, Premium supports up to 4,000, and purge handles smaller batches caused by large messages.
+
 ## 7.15.1 (Unreleased)
 
 ### Features Added

sdk/servicebus/azure-servicebus/api.md

@@ -25,6 +25,18 @@ namespace azure.servicebus
             ) -> None: ...
 
 
+    class azure.servicebus.DeleteMessagesResult:
+        property deleted_message_count: int    # Read-only
+
+        def __init__(self, deleted_message_count: int) -> None: ...
+
+
+    class azure.servicebus.PurgeMessagesResult(DeleteMessagesResult):
+        property deleted_message_count: int    # Read-only
+
+        def __init__(self, deleted_message_count: int) -> None: ...
+
+
     class azure.servicebus.ServiceBusClient: implements ContextManager 
         fully_qualified_namespace: str
 
@@ -366,6 +378,14 @@ namespace azure.servicebus
 
         def defer_message(self, message: ServiceBusReceivedMessage) -> None: ...
 
+        def delete_messages(
+                self, 
+                message_count: int, 
+                *, 
+                before_enqueued_time: Optional[datetime] = ..., 
+                timeout: Optional[float] = ...
+            ) -> DeleteMessagesResult: ...
+
         def peek_messages(
                 self, 
                 max_message_count: int = 1, 
@@ -375,6 +395,14 @@ namespace azure.servicebus
                 **kwargs: Any
             ) -> List[ServiceBusReceivedMessage]: ...
 
+        def purge_messages(
+                self, 
+                *, 
+                before_enqueued_time: Optional[datetime] = ..., 
+                max_message_count_per_batch: int = 500, 
+                timeout: Optional[float] = ...
+            ) -> PurgeMessagesResult: ...
+
         def receive_deferred_messages(
                 self, 
                 sequence_numbers: Union[int, List[int]], 
@@ -679,6 +707,14 @@ namespace azure.servicebus.aio
 
         async def defer_message(self, message: ServiceBusReceivedMessage) -> None: ...
 
+        async def delete_messages(
+                self, 
+                message_count: int, 
+                *, 
+                before_enqueued_time: Optional[datetime] = ..., 
+                timeout: Optional[float] = ...
+            ) -> DeleteMessagesResult: ...
+
         async def peek_messages(
                 self, 
                 max_message_count: int = 1, 
@@ -688,6 +724,14 @@ namespace azure.servicebus.aio
                 **kwargs: Any
             ) -> List[ServiceBusReceivedMessage]: ...
 
+        async def purge_messages(
+                self, 
+                *, 
+                before_enqueued_time: Optional[datetime] = ..., 
+                max_message_count_per_batch: int = 500, 
+                timeout: Optional[float] = ...
+            ) -> PurgeMessagesResult: ...
+
         async def receive_deferred_messages(
                 self, 
                 sequence_numbers: Union[int, List[int]], 

sdk/servicebus/azure-servicebus/api.metadata.yml

@@ -1,4 +1,4 @@
-apiMdSha256: a5f9da8d7980799f054d782f99eabf9b190925718f2bc4bdd86f01752a5c1bf8
-packageVersion: 7.15.0
+apiMdSha256: 46c52fca7868fd19d0ec06b18161ba148cd8496df1ea6bdf1600296afa83a7c7
+packageVersion: 7.16.0b1
 parserVersion: 0.3.31
-pythonVersion: 3.14.0
+pythonVersion: 3.11.14

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

@@ -11,6 +11,7 @@
 from ._servicebus_sender import ServiceBusSender
 from ._servicebus_receiver import ServiceBusReceiver
 from ._servicebus_session import ServiceBusSession
+from ._models import DeleteMessagesResult, PurgeMessagesResult
 from ._common.message import (
     ServiceBusMessage,
     ServiceBusMessageBatch,
@@ -43,6 +44,8 @@
     "ServiceBusReceiver",
     "ServiceBusSession",
     "ServiceBusSender",
+    "DeleteMessagesResult",
+    "PurgeMessagesResult",
     "TransportType",
     "AutoLockRenewer",
     "parse_connection_string",

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

@@ -384,6 +384,9 @@ def _do_retryable_operation(  # pylint: disable=inconsistent-return-statements
     ) -> Any:
         require_last_exception = kwargs.pop("require_last_exception", False)
         operation_requires_timeout = kwargs.pop("operation_requires_timeout", False)
+        suppress_next_session_timeout_message = kwargs.pop(
+            "suppress_next_session_timeout_message", False
+        )
         # Opt-in per call site so long-poll operations stay bounded only by the caller.
         apply_try_timeout = kwargs.pop("apply_try_timeout", False)
         try_timeout = self._config.try_timeout if apply_try_timeout else None
@@ -420,7 +423,10 @@ def _do_retryable_operation(  # pylint: disable=inconsistent-return-statements
                         # Keep the original message: it identifies which phase timed out.
                         # The session hint only applies to NEXT_AVAILABLE_SESSION receivers.
                         description = str(last_exception)
-                        if getattr(self, "_session_id", None) == NEXT_AVAILABLE_SESSION:
+                        if (
+                            getattr(self, "_session_id", None) == NEXT_AVAILABLE_SESSION
+                            and not suppress_next_session_timeout_message
+                        ):
                             description += (
                                 " If trying to receive from NEXT_AVAILABLE_SESSION, use max_wait_time"
                                 " on the ServiceBusReceiver to control the timeout."
@@ -434,6 +440,9 @@ def _do_retryable_operation(  # pylint: disable=inconsistent-return-statements
                     retried_times=retried_times,
                     last_exception=last_exception,
                     abs_timeout_time=abs_timeout_time,
+                    suppress_next_session_timeout_message=(
+                        suppress_next_session_timeout_message
+                    ),
                 )
 
     def _backoff(
@@ -442,6 +451,7 @@ def _backoff(
         last_exception: Exception,
         abs_timeout_time: Optional[float] = None,
         entity_name: Optional[str] = None,
+        suppress_next_session_timeout_message: bool = False,
     ) -> None:
         entity_name = entity_name or self._container_id
         backoff = _get_backoff_time(
@@ -468,7 +478,10 @@ def _backoff(
             if isinstance(last_exception, OperationTimeoutError):
                 # Keep the original message: it identifies which phase timed out.
                 description = str(last_exception)
-                if getattr(self, "_session_id", None) == NEXT_AVAILABLE_SESSION:
+                if (
+                    getattr(self, "_session_id", None) == NEXT_AVAILABLE_SESSION
+                    and not suppress_next_session_timeout_message
+                ):
                     description += (
                         " If trying to receive from NEXT_AVAILABLE_SESSION, use max_wait_time"
                         " on the ServiceBusReceiver to control the timeout."
@@ -570,8 +583,64 @@ def _mgmt_request_response_with_retry(
     def _open(self, timeout: Optional[float] = None):
         raise ValueError("Subclass should override the method.")
 
-    def _open_with_retry(self):
-        return self._do_retryable_operation(self._open, operation_requires_timeout=True, apply_try_timeout=True)
+    def _open_with_timeout(self, timeout: float):
+        return self._open(timeout=timeout)
+
+    def _open_with_retry(
+        self,
+        timeout: Optional[float] = None,
+        *,
+        suppress_next_session_timeout_message: bool = False,
+    ):
+        def open_with_timeout(timeout: Optional[float] = None):
+            if timeout is not None and timeout <= 0:
+                raise OperationTimeoutError()
+            if timeout is None:
+                return self._open()
+            return self._open_with_timeout(timeout)
+
+        return self._do_retryable_operation(
+            open_with_timeout,
+            timeout=timeout,
+            operation_requires_timeout=True,
+            apply_try_timeout=True,
+            suppress_next_session_timeout_message=(
+                suppress_next_session_timeout_message
+            ),
+        )
+
+    def _open_mgmt_link_with_retry(
+        self,
+        timeout: Optional[float] = None,
+        *,
+        suppress_next_session_timeout_message: bool = False,
+    ):
+        def open_mgmt_link(timeout: Optional[float] = None):
+            if timeout is not None and timeout <= 0:
+                raise OperationTimeoutError()
+            start_time = time.monotonic()
+            if timeout is None:
+                self._open()
+            else:
+                self._open_with_timeout(timeout)
+                timeout -= time.monotonic() - start_time
+                if timeout <= 0:
+                    raise OperationTimeoutError()
+            return self._amqp_transport.mgmt_client_setup(
+                self._handler,
+                node=self._mgmt_target.encode(self._config.encoding),
+                timeout=timeout,
+            )
+
+        return self._do_retryable_operation(
+            open_mgmt_link,
+            timeout=timeout,
+            operation_requires_timeout=True,
+            apply_try_timeout=True,
+            suppress_next_session_timeout_message=(
+                suppress_next_session_timeout_message
+            ),
+        )
 
     def _close_handler(self):
         if self._handler:

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

@@ -52,6 +52,7 @@
 REQUEST_RESPONSE_ADD_RULE_OPERATION = VENDOR + b":add-rule"
 REQUEST_RESPONSE_REMOVE_RULE_OPERATION = VENDOR + b":remove-rule"
 REQUEST_RESPONSE_GET_RULES_OPERATION = VENDOR + b":enumerate-rules"
+REQUEST_RESPONSE_BATCH_DELETE_MESSAGES_OPERATION = VENDOR + b":batch-delete-messages"
 
 CONTAINER_PREFIX = "servicebus.pysdk-"
 JWT_TOKEN_SCOPE = "https://servicebus.azure.net//.default"
@@ -84,6 +85,8 @@
 MGMT_REQUEST_RECEIVER_SETTLE_MODE = "receiver-settle-mode"
 MGMT_REQUEST_FROM_SEQUENCE_NUMBER = "from-sequence-number"
 MGMT_REQUEST_MAX_MESSAGE_COUNT = "message-count"
+MGMT_REQUEST_MESSAGE_COUNT = MGMT_REQUEST_MAX_MESSAGE_COUNT
+MGMT_REQUEST_ENQUEUED_TIME_UTC = "enqueued-time-utc"
 MGMT_REQUEST_MESSAGE = "message"
 MGMT_REQUEST_MESSAGES = "messages"
 MGMT_REQUEST_MESSAGE_ID = "message-id"

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

@@ -69,6 +69,25 @@ def peek_op(  # pylint: disable=inconsistent-return-statements
     )
 
 
+def batch_delete_op(  # pylint: disable=inconsistent-return-statements
+    status_code, message, description, amqp_transport, max_message_count
+):
+    condition = message.application_properties.get(MGMT_RESPONSE_MESSAGE_ERROR_CONDITION)
+    if status_code == 200 or (status_code == 404 and condition == ERROR_CODE_MESSAGE_NOT_FOUND):
+        deleted_count = message.value.get(b"message-count") if isinstance(message.value, dict) else None
+        if (
+            isinstance(deleted_count, bool)
+            or not isinstance(deleted_count, int)
+            or deleted_count < 0
+            or deleted_count > max_message_count
+        ):
+            raise ValueError("Batch delete response did not contain a valid message-count.")
+        return deleted_count
+    amqp_transport.handle_amqp_mgmt_error(
+        _LOGGER, "Batch delete messages failed.", condition, description, status_code
+    )
+
+
 def list_sessions_op(  # pylint: disable=inconsistent-return-statements
     status_code, message, description, amqp_transport
 ):

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

@@ -74,6 +74,16 @@
 # Number of seconds between the Unix epoch (1/1/1970) and year 1 CE.
 # This is the lowest value that can be represented by an AMQP timestamp.
 CE_ZERO_SECONDS: int = -62_135_596_800
+EPOCH_UTC: datetime.datetime = datetime.datetime(1970, 1, 1, tzinfo=TZ_UTC)
+
+
+def datetime_to_timestamp_ms(value: datetime.datetime) -> int:
+    if value.tzinfo is None:
+        normalized = value.replace(tzinfo=TZ_UTC)
+    else:
+        normalized = value.astimezone(TZ_UTC)
+    return (normalized - EPOCH_UTC) // datetime.timedelta(milliseconds=1)
+
 
 def utc_from_timestamp(timestamp: float) -> datetime.datetime:
     """

sdk/servicebus/azure-servicebus/azure/servicebus/_models.py

@@ -0,0 +1,24 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+
+
+class DeleteMessagesResult:
+    """The result of a batch delete operation."""
+
+    def __init__(self, deleted_message_count: int) -> None:
+        self._deleted_message_count = deleted_message_count
+
+    @property
+    def deleted_message_count(self) -> int:
+        """The number of messages deleted by the service.
+
+        :rtype: int
+        """
+        return self._deleted_message_count
+
+
+class PurgeMessagesResult(DeleteMessagesResult):
+    """The result of a purge operation, reporting the total number of messages deleted."""