[Service Bus] Add batch delete and purge APIs

[Eldert Grootenboer (EldertGrootenboer)]

Description

Azure Service Bus exposes the com.microsoft:batch-delete-messages management operation. This PR adds synchronous and asynchronous receiver-level batch-delete and purge APIs to azure-servicebus for regular, session, and subqueue receivers.

delete_messages(message_count, *, before_enqueued_time=None, timeout=None)
purge_messages(*, before_enqueued_time=None, max_message_count_per_batch=500, timeout=None)

Both methods return result objects containing deleted_message_count. Purge captures one enqueue-time cutoff and one operation-wide timeout, defaults to 500 messages per request, continues after every positive result, and stops only when the service returns zero. Callers can select another positive signed 32-bit request size; Service Bus enforces the current tier limit of 500 for Basic and Standard and 4,000 for Premium.

Management-link setup is retryable, but a destructive request is dispatched only once. Missing, malformed, negative, or excessive response counts fail instead of being interpreted as successful exhaustion. Samples cover default purge, Premium sizing, sessions, enqueue-time thresholds, returned counts, and partial or unknown outcomes.

Companion implementations: JavaScript, .NET, Java, and Go.

Testing: the full package suite completed with 360 passed and 271 skipped. The final worktree-bound batch-delete suite passes 44 tests. api.md and api.metadata.yml were regenerated from the local package.

All SDK Contribution checklist

• Public API additions are recorded in the generated API artifacts.
• CHANGELOG is updated.
• I have read the contribution guidelines.
• Pull request includes sync and async test coverage.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
9 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

🟡 Changes recommended

Management-link setup remains outside the promised retry and timeout guarantees, and Premium-only samples run unconditionally against Standard resources.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds receiver-level batch delete and purge support to Azure Service Bus, including synchronous and asynchronous APIs.

Changes:

• Adds delete/purge APIs and result models.
• Implements AMQP request handling, validation, cutoffs, and timeout budgets.
• Adds samples, tests, changelog, and API artifacts.

File summaries

File	Description
tests/unittests/test_batch_delete.py	Tests protocol, validation, sessions, timeouts, and purge loops.
samples/sync_samples/sample_code_servicebus.py	Adds synchronous usage examples.
samples/async_samples/sample_code_servicebus_async.py	Adds asynchronous usage examples.
CHANGELOG.md	Documents the new APIs.
azure/servicebus/aio/_servicebus_receiver_async.py	Implements asynchronous delete and purge.
azure/servicebus/aio/_base_handler_async.py	Adds timeout-aware asynchronous opening.
azure/servicebus/_servicebus_receiver.py	Implements synchronous delete and purge.
azure/servicebus/_models.py	Defines result objects.
azure/servicebus/_common/utils.py	Adds datetime-to-milliseconds conversion.
azure/servicebus/_common/mgmt_handlers.py	Parses and validates delete responses.
azure/servicebus/_common/constants.py	Adds management operation constants.
azure/servicebus/_base_handler.py	Adds timeout-aware synchronous opening.
azure/servicebus/__init__.py	Exports result types.
api.metadata.yml	Updates the API snapshot hash.
api.md	Records the new public API surface.

Review details

• Files reviewed: 15/15 changed files
• Comments generated: 6
• Review effort level: Balanced

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🟡 Changes recommended

Management-link setup cannot reliably recover from attach failures, and uAMQP bypasses the promised setup retry.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (2) — in code that hasn't changed since the last review.

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_client_async.py:379

• The async vendored pyamqp change is not mirrored in Event Hubs (sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py:345-392). The two vendored stacks are kept byte-identical; this introduces transport behavior drift and complicates future shared fixes. Apply the same async management-link refactor to the Event Hubs copy.
  sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/client.py:466
• This vendored pyamqp client change is not mirrored in the Event Hubs copy (sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/client.py:432-478). These stacks are maintained as byte-identical copies; leaving only Service Bus updated creates divergent timeout, cleanup, and management-link behavior that makes future shared fixes unsafe. Apply the equivalent change to the Event Hubs vendored client.

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_client_async.py:406

• asyncio.CancelledError is not an Exception, so cancellation during either readiness loop skips this cleanup and leaves the management link cached in an indeterminate state. Catch BaseException for this cleanup boundary and immediately re-raise, as the method already does.

        except Exception:

• Files reviewed: 23/23 changed files
• Comments generated: 4
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Timeout handling, response validation, and Event Hubs parity issues remain unresolved.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (3) — in code that hasn't changed since the last review.

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

• When pyAMQP management-link setup exhausts its budget, open_mgmt_link() raises the transport's built-in TimeoutError. Unlike _mgmt_request_response() (lines 544-546), this setup path does not translate that error before _do_retryable_operation() calls _handle_exception(), so callers receive a generic ServiceBusError instead of OperationTimeoutError. Translate the transport timeout here to preserve the public timeout contract.
  sdk/servicebus/azure-servicebus/azure/servicebus/_common/mgmt_handlers.py:87
• Returning zero for a 204 treats a response with no message-count as successful exhaustion. This contradicts the PR contract that missing counts fail and purge stops only when the service explicitly returns zero; a bodyless response can therefore terminate purge while eligible messages remain. Reject this response unless it contains a validated count, and update the test that currently asserts 204 == 0.
  sdk/servicebus/azure-servicebus/azure/servicebus/aio/_base_handler_async.py:464
• When async pyAMQP management-link setup exhausts its budget, open_mgmt_link_async() raises the transport's built-in TimeoutError. Unlike _mgmt_request_response() (lines 391-393), this setup path does not translate that error before _do_retryable_operation() calls _handle_exception(), so callers receive a generic ServiceBusError instead of OperationTimeoutError. Translate the transport timeout here to preserve the public timeout contract.

• Files reviewed: 23/23 changed files
• Comments generated: 2
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Response validation and sync/async timeout-budgeting issues remain unresolved.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (1)

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

• Add a second blank line between these top-level functions. The same module consistently leaves two blank lines between definitions (for example, lines 97–100), and Black will otherwise reformat this file.

    return (normalized - EPOCH_UTC) // datetime.timedelta(milliseconds=1)

• Files reviewed: 23/23 changed files
• Comments generated: 3
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Async timeout handling can leak a coroutine, and the async session sample may wait indefinitely.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (1) — in code that hasn't changed since the last review.

sdk/servicebus/azure-servicebus/samples/async_samples/sample_code_servicebus_async.py:482

• This default invocation targets SERVICEBUS_SESSION_ID, but the preceding async session example creates only the literal "<your session id>" (lines 333-340), while the test deployment supplies a different generated ID. On a clean namespace, entering this receiver waits indefinitely for a session this script never created. Keep this session-specific example opt-in or seed the configured session before opening it.

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

• Add a second blank line after this new top-level helper. The surrounding module uses two blank lines between top-level functions (for example, lines 97–100), and Black will reformat this section otherwise.

    return (normalized - EPOCH_UTC) // datetime.timedelta(milliseconds=1)

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

• The PR promises support for subqueue receivers, but the new suite exercises only regular and session receivers; there is no sub_queue case. Because dead-letter/transfer-dead-letter receivers use a different entity management target, add sync and async coverage that creates those receiver variants and verifies the batch-delete request is sent to the subqueue target.

    def delete_messages(
        self,
        message_count: int,
        *,
        before_enqueued_time: Optional[datetime.datetime] = None,

• Files reviewed: 23/23 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Setup timeout conversion, async session sample seeding, and formatting issues remain unresolved.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (3) — in code that hasn't changed since the last review.

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

• Only one blank line separates this new top-level helper from utc_from_timestamp, while this file uses two between top-level functions (for example, lines 97–100). This violates the Black/PEP 8 layout and will fail the formatting check; add the second blank line.
  sdk/servicebus/azure-servicebus/azure/servicebus/_transport/_pyamqp_transport.py:1093
• open_mgmt_link() raises the transport's raw TimeoutError, but this setup call runs directly inside _do_retryable_operation, not _mgmt_request_response (which normally converts transport timeouts). _handle_exception consequently wraps it as a non-retryable generic ServiceBusError, closes the handler, and aborts setup instead of surfacing OperationTimeoutError. Convert the transport timeout here before it reaches the retry helper.
  sdk/servicebus/azure-servicebus/samples/async_samples/sample_code_servicebus_async.py:450
• This executable sample opens SERVICEBUS_SESSION_ID, but the preceding example_session_ops_async() seeds only the distinct literal "<your session id>". On a clean sample namespace there is therefore no active message/session for this ID, so the exact-session receiver can wait instead of demonstrating purge. Seed this session inside the example (as the sync flow effectively does) so the snippet also works independently.

• Files reviewed: 23/23 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

The broad destructive API and transport changes require final human review, and the formatting nit remains unresolved.

Review details

Suppressed comments (1)

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

• Add the second module-level blank line before utc_from_timestamp; Black requires two blank lines between top-level function definitions, so this file will fail formatting validation.

    return (normalized - EPOCH_UTC) // datetime.timedelta(milliseconds=1)

• Files reviewed: 23/23 changed files
• Comments generated: 0 new
• Review effort level: Balanced