diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index afe2fa018c4d..f16eabf8f771 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -11,6 +11,7 @@ - Fixed a bug where messages returned by `receive_deferred_messages` had a `lock_token` of `None`, which prevented settling (completing, abandoning, dead-lettering, deferring) or renewing the lock on a deferred message in `PEEK_LOCK` mode. The lock token is now read from the `lock-token` field of the management-link response for deferred messages. ([#42454](https://github.com/Azure/azure-sdk-for-python/issues/42454)) - Read `com.microsoft:max-message-batch-size` vendor property from the AMQP sender link to correctly limit batch size on Premium large-message entities, where `max-message-size` can be up to 100 MB but the batch limit is 1 MB. - Fixed a bug where sending a batched or multi-message payload with `uamqp_transport=True` raised `TypeError: 'BatchMessage' object is not subscriptable` (and a masked `AttributeError` on the list path) when the first message carried a `message_id`, `session_id`, or `partition_key`. The batch envelope properties are now set through the transport-appropriate code path. (regression from [#42598](https://github.com/Azure/azure-sdk-for-python/pull/42598)) +- Fixed a bug where the async receiver factory methods on `azure.servicebus.aio.ServiceBusClient` (`get_queue_receiver`, `get_subscription_receiver`) and the async `ServiceBusReceiver` annotated the `auto_lock_renewer` keyword with the synchronous `AutoLockRenewer`, causing static type checkers to reject the documented `azure.servicebus.aio.AutoLockRenewer` usage. The annotation now references the async `AutoLockRenewer`, matching the docstrings and runtime behavior. ([#47948](https://github.com/Azure/azure-sdk-for-python/issues/47948)) ## 7.14.3 (2025-11-11) diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py b/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py index 03d06900dd29..b0387f27e4c3 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py @@ -21,7 +21,6 @@ from ._servicebus_sender_async import ServiceBusSender from ._servicebus_receiver_async import ServiceBusReceiver from .._common._configuration import Configuration -from .._common.auto_lock_renewer import AutoLockRenewer from .._common.utils import generate_dead_letter_entity_name, strip_protocol_from_uri from .._common.constants import ( ServiceBusSubQueue, @@ -32,6 +31,7 @@ if TYPE_CHECKING: from azure.core.credentials_async import AsyncTokenCredential + from ._async_auto_lock_renewer import AutoLockRenewer NextAvailableSessionType = Literal[ServiceBusSessionFilter.NEXT_AVAILABLE] @@ -352,7 +352,7 @@ def get_queue_receiver( sub_queue: Optional[Union[ServiceBusSubQueue, str]] = None, receive_mode: Union[ServiceBusReceiveMode, str] = ServiceBusReceiveMode.PEEK_LOCK, max_wait_time: Optional[float] = None, - auto_lock_renewer: Optional[AutoLockRenewer] = None, + auto_lock_renewer: Optional["AutoLockRenewer"] = None, prefetch_count: int = 0, **kwargs: Any, ) -> ServiceBusReceiver: @@ -539,7 +539,7 @@ def get_subscription_receiver( sub_queue: Optional[Union[ServiceBusSubQueue, str]] = None, receive_mode: Union[ServiceBusReceiveMode, str] = ServiceBusReceiveMode.PEEK_LOCK, max_wait_time: Optional[float] = None, - auto_lock_renewer: Optional[AutoLockRenewer] = None, + auto_lock_renewer: Optional["AutoLockRenewer"] = None, prefetch_count: int = 0, client_identifier: Optional[str] = None, socket_timeout: Optional[float] = None, diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py b/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py index fa9ba1a0d679..795e6a5b1d04 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py @@ -65,7 +65,7 @@ from .._pyamqp.aio._authentication_async import JWTTokenAuthAsync as pyamqp_JWTTokenAuthAsync from azure.core.credentials_async import AsyncTokenCredential from azure.core.credentials import AzureSasCredential, AzureNamedKeyCredential - from .._common.auto_lock_renewer import AutoLockRenewer + from ._async_auto_lock_renewer import AutoLockRenewer _LOGGER = logging.getLogger(__name__) diff --git a/sdk/servicebus/azure-servicebus/samples/async_samples/auto_lock_renew_async.py b/sdk/servicebus/azure-servicebus/samples/async_samples/auto_lock_renew_async.py index 12d3c9c3c8f6..2a328c77b25a 100644 --- a/sdk/servicebus/azure-servicebus/samples/async_samples/auto_lock_renew_async.py +++ b/sdk/servicebus/azure-servicebus/samples/async_samples/auto_lock_renew_async.py @@ -9,6 +9,8 @@ Example to show usage of AutoLockRenewer asynchronously: 1. Automatically renew locks on messages received from non-sessionful entity 2. Automatically renew locks on the session of sessionful entity + 3. Automatically renew locks by passing the AutoLockRenewer to the receiver factory + via the ``auto_lock_renewer`` keyword We do not guarantee that this SDK is thread-safe. We do not recommend reusing the ServiceBusClient, ServiceBusSender, ServiceBusReceiver across threads. It is up to the running @@ -26,6 +28,8 @@ FULLY_QUALIFIED_NAMESPACE = os.environ["SERVICEBUS_FULLY_QUALIFIED_NAMESPACE"] QUEUE_NAME = os.environ["SERVICEBUS_QUEUE_NAME"] SESSION_QUEUE_NAME = os.environ["SERVICEBUS_SESSION_QUEUE_NAME"] +TOPIC_NAME = os.environ["SERVICEBUS_TOPIC_NAME"] +SUBSCRIPTION_NAME = os.environ["SERVICEBUS_SUBSCRIPTION_NAME"] async def renew_lock_on_message_received_from_non_sessionful_entity(): @@ -58,6 +62,57 @@ async def renew_lock_on_message_received_from_non_sessionful_entity(): await renewer.close() +async def renew_lock_via_receiver_auto_lock_renewer_keyword() -> None: + # The `-> None` return annotation is intentional: it makes mypy type-check this + # function body (untyped bodies are skipped), so the auto_lock_renewer= calls below + # act as a static regression guard for both receiver-factory annotations + # (get_queue_receiver and get_subscription_receiver). + credential = DefaultAzureCredential() + servicebus_client = ServiceBusClient(FULLY_QUALIFIED_NAMESPACE, credential) + + async with servicebus_client: + async with servicebus_client.get_queue_sender(queue_name=QUEUE_NAME) as sender: + await sender.send_messages(ServiceBusMessage("message")) + print("Send message to non-sessionful queue.") + + # Pass the AutoLockRenewer directly to the receiver factory via the auto_lock_renewer + # keyword. Received messages are then automatically registered for lock renewal on + # receipt, without a manual renewer.register(...) call. The `async with` form ensures + # the renewer is shut down even if the receiver block raises. + async with AutoLockRenewer() as renewer: + async with servicebus_client.get_queue_receiver( + queue_name=QUEUE_NAME, auto_lock_renewer=renewer, prefetch_count=10 + ) as receiver: + received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5) + + await asyncio.sleep(100) # message handling for long period (E.g. application logic) + + for msg in received_msgs: + await receiver.complete_message(msg) + print("Complete queue messages.") + + async with servicebus_client.get_topic_sender(topic_name=TOPIC_NAME) as sender: + await sender.send_messages(ServiceBusMessage("message")) + print("Send message to topic.") + + # The auto_lock_renewer keyword behaves the same on the subscription receiver factory, + # so the type check also covers the get_subscription_receiver annotation from issue #47948. + async with AutoLockRenewer() as renewer: + async with servicebus_client.get_subscription_receiver( + topic_name=TOPIC_NAME, + subscription_name=SUBSCRIPTION_NAME, + auto_lock_renewer=renewer, + prefetch_count=10, + ) as receiver: + received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5) + + await asyncio.sleep(100) # message handling for long period (E.g. application logic) + + for msg in received_msgs: + await receiver.complete_message(msg) + print("Complete subscription messages.") + + async def renew_lock_on_session_of_the_sessionful_entity(): credential = DefaultAzureCredential() servicebus_client = ServiceBusClient(FULLY_QUALIFIED_NAMESPACE, credential) @@ -134,5 +189,6 @@ async def on_lock_renew_failure_callback(renewable, error): asyncio.run(renew_lock_on_message_received_from_non_sessionful_entity()) +asyncio.run(renew_lock_via_receiver_auto_lock_renewer_keyword()) asyncio.run(renew_lock_on_session_of_the_sessionful_entity()) asyncio.run(renew_lock_with_lock_renewal_failure_callback())