Skip to content

Commit 637bc8e

Browse files
committed
Merge branch 'master' of https://github.com/Azure/azure-sdk-for-python into fix_test_errors_dl_recognize_entities
* 'master' of https://github.com/Azure/azure-sdk-for-python: [ServiceBus] Settle non-deferred message through receiver link (Azure#10800) Add sync/async samples to demonstrate consuming from a number of sessions at one time. (Azure#11001) fixed alternative document input samples (Azure#11078) Fix pip link in azure-keyvault-secrets readme (Azure#11056) [ServiceBus] Update for readme and sample (Azure#11047)
2 parents a9a9a63 + 91d96a8 commit 637bc8e

File tree

13 files changed

+334
-78
lines changed

13 files changed

+334
-78
lines changed

sdk/keyvault/azure-keyvault-secrets/README.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -415,6 +415,7 @@ additional questions or comments.
415415
[recover_purge_sample]: https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/keyvault/azure-keyvault-secrets/samples/recover_purge_operations.py
416416
[recover_purge_async_sample]: https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/keyvault/azure-keyvault-secrets/samples/recover_purge_operations_async.py
417417
[keyvault_docs]: https://docs.microsoft.com/en-us/azure/key-vault/
418+
[pip]: https://pypi.org/project/pip/
418419
[pypi_package_secrets]: https://pypi.org/project/azure-keyvault-secrets/
419420
[reference_docs]: https://aka.ms/azsdk-python-keyvault-secrets-ref
420421
[secret_client_src]: https://github.com/Azure/azure-sdk-for-python/tree/master/sdk/keyvault/azure-keyvault-secrets/azure/keyvault/secrets

sdk/servicebus/azure-servicebus/README.md

Lines changed: 40 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ publish/subscribe capabilities, and the ability to easily scale as your needs gr
77

88
Use the Service Bus client library for Python to communicate between applications and services and implement asynchronous messaging patterns.
99

10-
* Create Service Bus namespaces, queues, topics, and subscriptions, and modify their settings
10+
* Create Service Bus namespaces, queues, topics, and subscriptions, and modify their settings.
1111
* Send and receive messages within your Service Bus channels.
1212
* Utilize message locks, sessions, and dead letter functionality to implement complex messaging patterns.
1313

@@ -29,7 +29,7 @@ pip install azure-servicebus --pre
2929
To use this package, you must have:
3030
* Azure subscription - [Create a free account][azure_sub]
3131
* Azure Service Bus - [Namespace and management credentials][service_bus_namespace]
32-
* Python 2.7, 3.5, 3.6, 3.7 or 3.8 - [Install Python][python]
32+
* Python 2.7, 3.5 or later - [Install Python][python]
3333

3434

3535
If you need an Azure service bus namespace, you can create it via the [Azure Portal][azure_namespace_creation].
@@ -43,19 +43,17 @@ az servicebus namespace create --resource-group <resource-group-name> --name <se
4343

4444
Interaction with Service Bus starts with an instance of the `ServiceBusClient` class. You either need a **connection string with SAS key**, or a **namespace** and one of its **account keys** to instantiate the client object.
4545

46-
#### Get credentials
46+
#### Create client from connection string
4747

48-
Use the [Azure CLI][azure_cli] snippet below to populate an environment variable with the service bus connection string (you can also find these values in the [Azure portal][azure_portal]. The snippet is formatted for the Bash shell.
48+
- Get credentials: Use the [Azure CLI][azure_cli] snippet below to populate an environment variable with the service bus connection string (you can also find these values in the [Azure Portal][azure_portal] by following the step-by-step guide to [Get a service bus connection string][get_servicebus_conn_str]). The snippet is formatted for the Bash shell.
4949

5050
```Bash
5151
RES_GROUP=<resource-group-name>
5252
NAMESPACE_NAME=<servicebus-namespace-name>
5353

54-
export SERVICE_BUS_CONN_STR=$(az servicebus namespace authorization-rule keys list --resource-group $RES_GROUP --namespace-name $NAMESPACE_NAME --query RootManageSharedAccessKey --output tsv)
54+
export SERVICE_BUS_CONN_STR=$(az servicebus namespace authorization-rule keys list --resource-group $RES_GROUP --namespace-name $NAMESPACE_NAME --name RootManageSharedAccessKey --query primaryConnectionString --output tsv)
5555
```
5656

57-
#### Create client
58-
5957
Once you've populated the `SERVICE_BUS_CONN_STR` environment variable, you can create the `ServiceBusClient`.
6058

6159
```Python
@@ -68,6 +66,28 @@ with ServiceBusClient.from_connection_string(connstr) as client:
6866
...
6967
```
7068

69+
#### Create client using the azure-identity library:
70+
71+
```python
72+
import os
73+
from azure.servicebus import ServiceBusClient
74+
from azure.identity import DefaultAzureCredential
75+
76+
credential = DefaultAzureCredential()
77+
78+
FULLY_QUALIFIED_NAMESPACE = os.environ['SERVICE_BUS_FULLY_QUALIFIED_NAMESPACE']
79+
with ServiceBusClient(FULLY_QUALIFIED_NAMESPACE, credential):
80+
...
81+
```
82+
83+
- This constructor takes the fully qualified namespace of your Service Bus instance and a credential that implements the
84+
[TokenCredential][token_credential_interface]
85+
protocol. There are implementations of the `TokenCredential` protocol available in the
86+
[azure-identity package][pypi_azure_identity]. The fully qualified namespace is of the format `<yournamespace.servicebus.windows.net>`.
87+
- When using Azure Active Directory, your principal must be assigned a role which allows access to Service Bus, such as the
88+
Azure Service Bus Data Owner role. For more information about using Azure Active Directory authorization with Service Bus,
89+
please refer to [the associated documentation][servicebus_aad_authentication].
90+
7191
Note: client can be initialized without a context manager, but must be manually closed via client.close() to not leak resources.
7292

7393
## Key concepts
@@ -88,22 +108,22 @@ To interact with these resources, one should be familiar with the following SDK
88108

89109
* [Sender](./azure/servicebus/_servicebus_sender.py): To send messages to a Queue or Topic, one would use the corresponding `get_queue_sender` or `get_topic_sender` method off of a `ServiceBusClient` instance as seen [here](./samples/sync_samples/send_queue.py).
90110

91-
* [Receiver](./azure/servicebus/_servicebus_receiver.py): To receive messages from a Queue or Subscription, one would use the corrosponding `get_queue_receiver` or `get_subscription_receiver` method off of a `ServiceBusClient` instance as seen [here](./samples/sync_samples/receive_queue.py).
111+
* [Receiver](./azure/servicebus/_servicebus_receiver.py): To receive messages from a Queue or Subscription, one would use the corresponding `get_queue_receiver` or `get_subscription_receiver` method off of a `ServiceBusClient` instance as seen [here](./samples/sync_samples/receive_queue.py).
92112

93113
* [Message](./azure/servicebus/_common/message.py): When sending, this is the type you will construct to contain your payload. When receiving, this is where you will access the payload and control how the message is "settled" (completed, dead-lettered, etc); these functions are only available on a received message.
94114

95115
## Examples
96116

97117
The following sections provide several code snippets covering some of the most common Service Bus tasks, including:
98118

99-
* [Send a message to a queue](#send-to-a-queue)
100-
* [Receive a message from a queue](#receive-from-a-queue)
101-
* [Defer a message on receipt](#defer-a-message)
119+
* [Send a message to a queue](#send-a-message-to-a-queue)
120+
* [Receive a message from a queue](#receive-a-message-from-a-queue)
121+
* [Defer a message on receipt](#defer-a-message-on-receipt)
102122

103123
To perform management tasks such as creating and deleting queues/topics/subscriptions, please utilize the azure-mgmt-servicebus library, available [here][servicebus_management_repository].
104124

105125

106-
### Send to a queue
126+
### Send a message to a queue
107127

108128
This example sends a message to a queue that is assumed to already exist, created via the Azure portal or az commands.
109129

@@ -121,7 +141,7 @@ with ServiceBusClient.from_connection_string(connstr) as client:
121141
sender.send(message)
122142
```
123143

124-
### Receive from a queue
144+
### Receive a message from a queue
125145

126146
To receive from a queue, you can either perform a one-off receive via "receiver.receive()" or receive persistently as follows:
127147

@@ -139,7 +159,7 @@ with ServiceBusClient.from_connection_string(connstr) as client:
139159
msg.complete()
140160
```
141161

142-
### Defer a message
162+
### Defer a message on receipt
143163

144164
When receiving from a queue, you have multiple actions you can take on the messages you receive. Where the prior example completes a message,
145165
permanently removing it from the queue and marking as complete, this example demonstrates how to defer the message, sending it back to the queue
@@ -150,6 +170,7 @@ from azure.servicebus import ServiceBusClient
150170

151171
import os
152172
connstr = os.environ['SERVICE_BUS_CONN_STR']
173+
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
153174

154175
with ServiceBusClient.from_connection_string(connstr) as client:
155176
with client.get_queue_receiver(queue_name) as receiver:
@@ -225,4 +246,8 @@ contact [[email protected]](mailto:[email protected]) with any additio
225246
[topic_concept]: https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview#topics
226247
[subscription_concept]: https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-queues-topics-subscriptions#topics-and-subscriptions
227248
[azure_namespace_creation]: https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-create-namespace-portal
228-
[servicebus_management_repository]: https://github.com/Azure/azure-sdk-for-python/tree/master/sdk/servicebus/azure-mgmt-servicebus
249+
[servicebus_management_repository]: https://github.com/Azure/azure-sdk-for-python/tree/master/sdk/servicebus/azure-mgmt-servicebus
250+
[get_servicebus_conn_str]: https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-create-namespace-portal#get-the-connection-string
251+
[servicebus_aad_authentication]: https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-authentication-and-authorization
252+
[token_credential_interface]: ../../core/azure-core/azure/core/credentials.py
253+
[pypi_azure_identity]: https://pypi.org/project/azure-identity/

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

Lines changed: 80 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,12 @@
66

77
import datetime
88
import uuid
9+
import functools
10+
import logging
911
from typing import Optional, List, Union, Generator
1012

1113
import uamqp
12-
from uamqp import types
14+
from uamqp import types, errors
1315

1416
from .constants import (
1517
_BATCH_MESSAGE_OVERHEAD_COST,
@@ -34,7 +36,8 @@
3436
MESSAGE_DEAD_LETTER,
3537
MESSAGE_ABANDON,
3638
MESSAGE_DEFER,
37-
MESSAGE_RENEW_LOCK
39+
MESSAGE_RENEW_LOCK,
40+
DEADLETTERNAME
3841
)
3942
from ..exceptions import (
4043
MessageAlreadySettled,
@@ -44,6 +47,8 @@
4447
)
4548
from .utils import utc_from_timestamp, utc_now
4649

50+
_LOGGER = logging.getLogger(__name__)
51+
4752

4853
class Message(object): # pylint: disable=too-many-public-methods,too-many-instance-attributes
4954
"""A Service Bus Message.
@@ -436,9 +441,10 @@ class ReceivedMessage(PeekMessage):
436441
:dedent: 4
437442
:caption: Checking the properties on a received message.
438443
"""
439-
def __init__(self, message, mode=ReceiveSettleMode.PeekLock):
444+
def __init__(self, message, mode=ReceiveSettleMode.PeekLock, **kwargs):
440445
super(ReceivedMessage, self).__init__(message=message)
441446
self._settled = (mode == ReceiveSettleMode.ReceiveAndDelete)
447+
self._is_deferred_message = kwargs.get("is_deferred_message", False)
442448
self.auto_renew_error = None
443449

444450
def _is_live(self, action):
@@ -458,6 +464,69 @@ def _is_live(self, action):
458464
except AttributeError:
459465
pass
460466

467+
def _settle_message(
468+
self,
469+
settle_operation,
470+
dead_letter_details=None
471+
):
472+
try:
473+
if not self._is_deferred_message:
474+
try:
475+
self._settle_via_receiver_link(settle_operation, dead_letter_details)()
476+
return
477+
except RuntimeError as exception:
478+
_LOGGER.info(
479+
"Message settling: %r has encountered an exception (%r)."
480+
"Trying to settle through management link",
481+
settle_operation,
482+
exception
483+
)
484+
self._settle_via_mgmt_link(settle_operation, dead_letter_details)()
485+
except Exception as e:
486+
raise MessageSettleFailed(settle_operation, e)
487+
488+
def _settle_via_mgmt_link(self, settle_operation, dead_letter_details=None):
489+
# pylint: disable=protected-access
490+
if settle_operation == MESSAGE_COMPLETE:
491+
return functools.partial(
492+
self._receiver._settle_message,
493+
SETTLEMENT_COMPLETE,
494+
[self.lock_token],
495+
)
496+
if settle_operation == MESSAGE_ABANDON:
497+
return functools.partial(
498+
self._receiver._settle_message,
499+
SETTLEMENT_ABANDON,
500+
[self.lock_token],
501+
)
502+
if settle_operation == MESSAGE_DEAD_LETTER:
503+
return functools.partial(
504+
self._receiver._settle_message,
505+
SETTLEMENT_DEADLETTER,
506+
[self.lock_token],
507+
dead_letter_details=dead_letter_details
508+
)
509+
if settle_operation == MESSAGE_DEFER:
510+
return functools.partial(
511+
self._receiver._settle_message,
512+
SETTLEMENT_DEFER,
513+
[self.lock_token],
514+
)
515+
raise ValueError("Unsupported settle operation type: {}".format(settle_operation))
516+
517+
def _settle_via_receiver_link(self, settle_operation, dead_letter_details=None):
518+
if settle_operation == MESSAGE_COMPLETE:
519+
return functools.partial(self.message.accept)
520+
if settle_operation == MESSAGE_ABANDON:
521+
return functools.partial(self.message.modify, True, False)
522+
if settle_operation == MESSAGE_DEAD_LETTER:
523+
# note: message.reject() can not set reason and description properly due to the issue
524+
# https://github.com/Azure/azure-uamqp-python/issues/155
525+
return functools.partial(self.message.reject, condition=DEADLETTERNAME)
526+
if settle_operation == MESSAGE_DEFER:
527+
return functools.partial(self.message.modify, True, True)
528+
raise ValueError("Unsupported settle operation type: {}".format(settle_operation))
529+
461530
@property
462531
def settled(self):
463532
# type: () -> bool
@@ -535,11 +604,9 @@ def complete(self):
535604
:raises: ~azure.servicebus.common.errors.SessionLockExpired if session lock has already expired.
536605
:raises: ~azure.servicebus.common.errors.MessageSettleFailed if message settle operation fails.
537606
"""
607+
# pylint: disable=protected-access
538608
self._is_live(MESSAGE_COMPLETE)
539-
try:
540-
self._receiver._settle_message(SETTLEMENT_COMPLETE, [self.lock_token]) # pylint: disable=protected-access
541-
except Exception as e:
542-
raise MessageSettleFailed(MESSAGE_COMPLETE, e)
609+
self._settle_message(MESSAGE_COMPLETE)
543610
self._settled = True
544611

545612
def dead_letter(self, reason=None, description=None):
@@ -560,17 +627,12 @@ def dead_letter(self, reason=None, description=None):
560627
"""
561628
# pylint: disable=protected-access
562629
self._is_live(MESSAGE_DEAD_LETTER)
630+
563631
details = {
564632
MGMT_REQUEST_DEAD_LETTER_REASON: str(reason) if reason else "",
565633
MGMT_REQUEST_DEAD_LETTER_DESCRIPTION: str(description) if description else ""}
566-
try:
567-
self._receiver._settle_message(
568-
SETTLEMENT_DEADLETTER,
569-
[self.lock_token],
570-
dead_letter_details=details
571-
)
572-
except Exception as e:
573-
raise MessageSettleFailed(MESSAGE_DEAD_LETTER, e)
634+
635+
self._settle_message(MESSAGE_DEAD_LETTER, dead_letter_details=details)
574636
self._settled = True
575637

576638
def abandon(self):
@@ -585,11 +647,9 @@ def abandon(self):
585647
:raises: ~azure.servicebus.common.errors.SessionLockExpired if session lock has already expired.
586648
:raises: ~azure.servicebus.common.errors.MessageSettleFailed if message settle operation fails.
587649
"""
650+
# pylint: disable=protected-access
588651
self._is_live(MESSAGE_ABANDON)
589-
try:
590-
self._receiver._settle_message(SETTLEMENT_ABANDON, [self.lock_token]) # pylint: disable=protected-access
591-
except Exception as e:
592-
raise MessageSettleFailed(MESSAGE_ABANDON, e)
652+
self._settle_message(MESSAGE_ABANDON)
593653
self._settled = True
594654

595655
def defer(self):
@@ -606,10 +666,7 @@ def defer(self):
606666
:raises: ~azure.servicebus.common.errors.MessageSettleFailed if message settle operation fails.
607667
"""
608668
self._is_live(MESSAGE_DEFER)
609-
try:
610-
self._receiver._settle_message(SETTLEMENT_DEFER, [self.lock_token]) # pylint: disable=protected-access
611-
except Exception as e:
612-
raise MessageSettleFailed(MESSAGE_DEFER, e)
669+
self._settle_message(MESSAGE_DEFER)
613670
self._settled = True
614671

615672
def renew_lock(self):

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ def deferred_message_op(
6969
parsed = []
7070
for m in message.get_data()[b'messages']:
7171
wrapped = uamqp.Message.decode_from_bytes(bytearray(m[b'message']))
72-
parsed.append(message_type(wrapped, mode))
72+
parsed.append(message_type(wrapped, mode, is_deferred_message=True))
7373
return parsed
7474
if status_code in [202, 204]:
7575
return []

0 commit comments

Comments
 (0)