1# ------------------------------------------------------------------------- 2# Copyright (c) Microsoft Corporation. All rights reserved. 3# Licensed under the MIT License. See License.txt in the project root for 4# license information. 5# -------------------------------------------------------------------------- 6from azure.common import ( 7 AzureConflictHttpError, 8 AzureHttpError, 9) 10 11from ..common._auth import ( 12 _StorageSASAuthentication, 13 _StorageSharedKeyAuthentication, 14) 15from ..common._common_conversion import ( 16 _int_to_str, 17 _to_str, 18) 19from ..common._connection import _ServiceParameters 20from ..common._constants import ( 21 SERVICE_HOST_BASE, 22 DEFAULT_PROTOCOL, 23) 24from ..common._deserialization import ( 25 _convert_xml_to_service_properties, 26 _convert_xml_to_signed_identifiers, 27 _convert_xml_to_service_stats, 28) 29from ..common._error import ( 30 _dont_fail_not_exist, 31 _dont_fail_on_exist, 32 _validate_not_none, 33 _ERROR_CONFLICT, 34 _ERROR_STORAGE_MISSING_INFO, 35 _validate_access_policies, 36 _validate_encryption_required, 37 _validate_decryption_required, 38) 39from ..common._http import ( 40 HTTPRequest, 41) 42from ..common._serialization import ( 43 _convert_signed_identifiers_to_xml, 44 _convert_service_properties_to_xml, 45) 46from ..common._serialization import ( 47 _get_request_body, 48 _add_metadata_headers, 49) 50from ..common.models import ( 51 Services, 52 ListGenerator, 53 _OperationContext, 54) 55from .sharedaccesssignature import ( 56 QueueSharedAccessSignature, 57) 58from ..common.storageclient import StorageClient 59from ._deserialization import ( 60 _convert_xml_to_queues, 61 _convert_xml_to_queue_messages, 62 _parse_queue_message_from_headers, 63 _parse_metadata_and_message_count, 64) 65from ._serialization import ( 66 _convert_queue_message_xml, 67 _get_path, 68) 69from .models import ( 70 QueueMessageFormat, 71) 72from ._constants import ( 73 X_MS_VERSION, 74 __version__ as package_version, 75) 76 77_QUEUE_ALREADY_EXISTS_ERROR_CODE = 'QueueAlreadyExists' 78_QUEUE_NOT_FOUND_ERROR_CODE = 'QueueNotFound' 79_HTTP_RESPONSE_NO_CONTENT = 204 80 81 82class QueueService(StorageClient): 83 ''' 84 This is the main class managing queue resources. 85 86 The Queue service stores messages. A queue can contain an unlimited number of 87 messages, each of which can be up to 64KB in size. Messages are generally added 88 to the end of the queue and retrieved from the front of the queue, although 89 first in, first out (FIFO) behavior is not guaranteed. 90 91 :ivar function(data) encode_function: 92 A function used to encode queue messages. Takes as 93 a parameter the data passed to the put_message API and returns the encoded 94 message. Defaults to take text and xml encode, but bytes and other 95 encodings can be used. For example, base64 may be preferable for developing 96 across multiple Azure Storage libraries in different languages. See the 97 :class:`~azure.storage.queue.models.QueueMessageFormat` for xml, base64 and 98 no encoding methods as well as binary equivalents. 99 :ivar function(data) decode_function: 100 A function used to encode decode messages. Takes as 101 a parameter the data returned by the get_messages and peek_messages APIs and 102 returns the decoded message. Defaults to return text and xml decode, but 103 bytes and other decodings can be used. For example, base64 may be preferable 104 for developing across multiple Azure Storage libraries in different languages. 105 See the :class:`~azure.storage.queue.models.QueueMessageFormat` for xml, base64 106 and no decoding methods as well as binary equivalents. 107 :ivar object key_encryption_key: 108 The key-encryption-key optionally provided by the user. If provided, will be used to 109 encrypt/decrypt in supported methods. 110 For methods requiring decryption, either the key_encryption_key OR the resolver must be provided. 111 If both are provided, the resolver will take precedence. 112 Must implement the following methods for APIs requiring encryption: 113 wrap_key(key)--wraps the specified key (bytes) using an algorithm of the user's choice. Returns the encrypted key as bytes. 114 get_key_wrap_algorithm()--returns the algorithm used to wrap the specified symmetric key. 115 get_kid()--returns a string key id for this key-encryption-key. 116 Must implement the following methods for APIs requiring decryption: 117 unwrap_key(key, algorithm)--returns the unwrapped form of the specified symmetric key using the string-specified algorithm. 118 get_kid()--returns a string key id for this key-encryption-key. 119 :ivar function key_resolver_function(kid): 120 A function to resolve keys optionally provided by the user. If provided, will be used to decrypt in supported methods. 121 For methods requiring decryption, either the key_encryption_key OR 122 the resolver must be provided. If both are provided, the resolver will take precedence. 123 It uses the kid string to return a key-encryption-key implementing the interface defined above. 124 :ivar bool require_encryption: 125 A flag that may be set to ensure that all messages successfully uploaded to the queue and all those downloaded and 126 successfully read from the queue are/were encrypted while on the server. If this flag is set, all required 127 parameters for encryption/decryption must be provided. See the above comments on the key_encryption_key and resolver. 128 ''' 129 130 def __init__(self, account_name=None, account_key=None, sas_token=None, is_emulated=False, 131 protocol=DEFAULT_PROTOCOL, endpoint_suffix=SERVICE_HOST_BASE, request_session=None, 132 connection_string=None, socket_timeout=None, token_credential=None): 133 ''' 134 :param str account_name: 135 The storage account name. This is used to authenticate requests 136 signed with an account key and to construct the storage endpoint. It 137 is required unless a connection string is given. 138 :param str account_key: 139 The storage account key. This is used for shared key authentication. 140 :param str sas_token: 141 A shared access signature token to use to authenticate requests 142 instead of the account key. If account key and sas token are both 143 specified, account key will be used to sign. 144 :param bool is_emulated: 145 Whether to use the emulator. Defaults to False. If specified, will 146 override all other parameters besides connection string and request 147 session. 148 :param str protocol: 149 The protocol to use for requests. Defaults to https. 150 :param str endpoint_suffix: 151 The host base component of the url, minus the account name. Defaults 152 to Azure (core.windows.net). Override this to use the China cloud 153 (core.chinacloudapi.cn). 154 :param requests.Session request_session: 155 The session object to use for http requests. 156 :param str connection_string: 157 If specified, this will override all other parameters besides 158 request session. See 159 http://azure.microsoft.com/en-us/documentation/articles/storage-configure-connection-string/ 160 for the connection string format. 161 :param int socket_timeout: 162 If specified, this will override the default socket timeout. The timeout specified is in seconds. 163 See DEFAULT_SOCKET_TIMEOUT in _constants.py for the default value. 164 :param token_credential: 165 A token credential used to authenticate HTTPS requests. The token value 166 should be updated before its expiration. 167 :type `~..common.TokenCredential` 168 ''' 169 service_params = _ServiceParameters.get_service_parameters( 170 'queue', 171 account_name=account_name, 172 account_key=account_key, 173 sas_token=sas_token, 174 token_credential=token_credential, 175 is_emulated=is_emulated, 176 protocol=protocol, 177 endpoint_suffix=endpoint_suffix, 178 request_session=request_session, 179 connection_string=connection_string, 180 socket_timeout=socket_timeout) 181 182 super(QueueService, self).__init__(service_params) 183 184 if self.account_key: 185 self.authentication = _StorageSharedKeyAuthentication( 186 self.account_name, 187 self.account_key, 188 self.is_emulated 189 ) 190 elif self.sas_token: 191 self.authentication = _StorageSASAuthentication(self.sas_token) 192 elif self.token_credential: 193 self.authentication = self.token_credential 194 else: 195 raise ValueError(_ERROR_STORAGE_MISSING_INFO) 196 197 self.encode_function = QueueMessageFormat.text_xmlencode 198 self.decode_function = QueueMessageFormat.text_xmldecode 199 self.key_encryption_key = None 200 self.key_resolver_function = None 201 self.require_encryption = False 202 self._X_MS_VERSION = X_MS_VERSION 203 self._update_user_agent_string(package_version) 204 205 def generate_account_shared_access_signature(self, resource_types, permission, 206 expiry, start=None, ip=None, protocol=None): 207 ''' 208 Generates a shared access signature for the queue service. 209 Use the returned signature with the sas_token parameter of QueueService. 210 211 :param ResourceTypes resource_types: 212 Specifies the resource types that are accessible with the account SAS. 213 :param AccountPermissions permission: 214 The permissions associated with the shared access signature. The 215 user is restricted to operations allowed by the permissions. 216 Required unless an id is given referencing a stored access policy 217 which contains this field. This field must be omitted if it has been 218 specified in an associated stored access policy. 219 :param expiry: 220 The time at which the shared access signature becomes invalid. 221 Required unless an id is given referencing a stored access policy 222 which contains this field. This field must be omitted if it has 223 been specified in an associated stored access policy. Azure will always 224 convert values to UTC. If a date is passed in without timezone info, it 225 is assumed to be UTC. 226 :type expiry: datetime or str 227 :param start: 228 The time at which the shared access signature becomes valid. If 229 omitted, start time for this call is assumed to be the time when the 230 storage service receives the request. Azure will always convert values 231 to UTC. If a date is passed in without timezone info, it is assumed to 232 be UTC. 233 :type start: datetime or str 234 :param str ip: 235 Specifies an IP address or a range of IP addresses from which to accept requests. 236 If the IP address from which the request originates does not match the IP address 237 or address range specified on the SAS token, the request is not authenticated. 238 For example, specifying sip=168.1.5.65 or sip=168.1.5.60-168.1.5.70 on the SAS 239 restricts the request to those IP addresses. 240 :param str protocol: 241 Specifies the protocol permitted for a request made. The default value 242 is https,http. See :class:`~..common.models.Protocol` for possible values. 243 :return: A Shared Access Signature (sas) token. 244 :rtype: str 245 ''' 246 _validate_not_none('self.account_name', self.account_name) 247 _validate_not_none('self.account_key', self.account_key) 248 249 sas = QueueSharedAccessSignature(self.account_name, self.account_key) 250 return sas.generate_account(Services.QUEUE, resource_types, permission, 251 expiry, start=start, ip=ip, protocol=protocol) 252 253 def generate_queue_shared_access_signature(self, queue_name, 254 permission=None, 255 expiry=None, 256 start=None, 257 id=None, 258 ip=None, protocol=None, ): 259 ''' 260 Generates a shared access signature for the queue. 261 Use the returned signature with the sas_token parameter of QueueService. 262 263 :param str queue_name: 264 The name of the queue to create a SAS token for. 265 :param QueuePermissions permission: 266 The permissions associated with the shared access signature. The 267 user is restricted to operations allowed by the permissions. 268 Required unless an id is given referencing a stored access policy 269 which contains this field. This field must be omitted if it has been 270 specified in an associated stored access policy. 271 :param expiry: 272 The time at which the shared access signature becomes invalid. 273 Required unless an id is given referencing a stored access policy 274 which contains this field. This field must be omitted if it has 275 been specified in an associated stored access policy. Azure will always 276 convert values to UTC. If a date is passed in without timezone info, it 277 is assumed to be UTC. 278 :type expiry: datetime or str 279 :param start: 280 The time at which the shared access signature becomes valid. If 281 omitted, start time for this call is assumed to be the time when the 282 storage service receives the request. Azure will always convert values 283 to UTC. If a date is passed in without timezone info, it is assumed to 284 be UTC. 285 :type start: datetime or str 286 :param str id: 287 A unique value up to 64 characters in length that correlates to a 288 stored access policy. To create a stored access policy, use :func:`~set_queue_acl`. 289 :param str ip: 290 Specifies an IP address or a range of IP addresses from which to accept requests. 291 If the IP address from which the request originates does not match the IP address 292 or address range specified on the SAS token, the request is not authenticated. 293 For example, specifying sip='168.1.5.65' or sip='168.1.5.60-168.1.5.70' on the SAS 294 restricts the request to those IP addresses. 295 :param str protocol: 296 Specifies the protocol permitted for a request made. The default value 297 is https,http. See :class:`~..common.models.Protocol` for possible values. 298 :return: A Shared Access Signature (sas) token. 299 :rtype: str 300 ''' 301 _validate_not_none('queue_name', queue_name) 302 _validate_not_none('self.account_name', self.account_name) 303 _validate_not_none('self.account_key', self.account_key) 304 305 sas = QueueSharedAccessSignature(self.account_name, self.account_key) 306 return sas.generate_queue( 307 queue_name, 308 permission=permission, 309 expiry=expiry, 310 start=start, 311 id=id, 312 ip=ip, 313 protocol=protocol, 314 ) 315 316 def get_queue_service_stats(self, timeout=None): 317 ''' 318 Retrieves statistics related to replication for the Queue service. It is 319 only available when read-access geo-redundant replication is enabled for 320 the storage account. 321 322 With geo-redundant replication, Azure Storage maintains your data durable 323 in two locations. In both locations, Azure Storage constantly maintains 324 multiple healthy replicas of your data. The location where you read, 325 create, update, or delete data is the primary storage account location. 326 The primary location exists in the region you choose at the time you 327 create an account via the Azure Management Azure classic portal, for 328 example, North Central US. The location to which your data is replicated 329 is the secondary location. The secondary location is automatically 330 determined based on the location of the primary; it is in a second data 331 center that resides in the same region as the primary location. Read-only 332 access is available from the secondary location, if read-access geo-redundant 333 replication is enabled for your storage account. 334 335 :param int timeout: 336 The timeout parameter is expressed in seconds. 337 :return: The queue service stats. 338 :rtype: :class:`~..common.models.ServiceStats` 339 ''' 340 request = HTTPRequest() 341 request.method = 'GET' 342 request.host_locations = self._get_host_locations(primary=False, secondary=True) 343 request.path = _get_path() 344 request.query = { 345 'restype': 'service', 346 'comp': 'stats', 347 'timeout': _int_to_str(timeout), 348 } 349 350 return self._perform_request(request, _convert_xml_to_service_stats) 351 352 def get_queue_service_properties(self, timeout=None): 353 ''' 354 Gets the properties of a storage account's Queue service, including 355 logging, analytics and CORS rules. 356 357 :param int timeout: 358 The server timeout, expressed in seconds. 359 :return: The queue service properties. 360 :rtype: :class:`~..common.models.ServiceProperties` 361 ''' 362 request = HTTPRequest() 363 request.method = 'GET' 364 request.host_locations = self._get_host_locations(secondary=True) 365 request.path = _get_path() 366 request.query = { 367 'restype': 'service', 368 'comp': 'properties', 369 'timeout': _int_to_str(timeout), 370 } 371 372 return self._perform_request(request, _convert_xml_to_service_properties) 373 374 def set_queue_service_properties(self, logging=None, hour_metrics=None, 375 minute_metrics=None, cors=None, timeout=None): 376 ''' 377 Sets the properties of a storage account's Queue service, including 378 Azure Storage Analytics. If an element (ex Logging) is left as None, the 379 existing settings on the service for that functionality are preserved. 380 For more information on Azure Storage Analytics, see 381 https://msdn.microsoft.com/en-us/library/azure/hh343270.aspx. 382 383 :param Logging logging: 384 The logging settings provide request logs. 385 :param Metrics hour_metrics: 386 The hour metrics settings provide a summary of request 387 statistics grouped by API in hourly aggregates for queuess. 388 :param Metrics minute_metrics: 389 The minute metrics settings provide request statistics 390 for each minute for queues. 391 :param cors: 392 You can include up to five CorsRule elements in the 393 list. If an empty list is specified, all CORS rules will be deleted, 394 and CORS will be disabled for the service. For detailed information 395 about CORS rules and evaluation logic, see 396 https://msdn.microsoft.com/en-us/library/azure/dn535601.aspx. 397 :type cors: list(:class:`~..common.models.CorsRule`) 398 :param int timeout: 399 The server timeout, expressed in seconds. 400 ''' 401 request = HTTPRequest() 402 request.method = 'PUT' 403 request.host_locations = self._get_host_locations() 404 request.path = _get_path() 405 request.query = { 406 'restype': 'service', 407 'comp': 'properties', 408 'timeout': _int_to_str(timeout), 409 } 410 request.body = _get_request_body( 411 _convert_service_properties_to_xml(logging, hour_metrics, minute_metrics, cors)) 412 self._perform_request(request) 413 414 def list_queues(self, prefix=None, num_results=None, include_metadata=False, 415 marker=None, timeout=None): 416 ''' 417 Returns a generator to list the queues. The generator will lazily follow 418 the continuation tokens returned by the service and stop when all queues 419 have been returned or num_results is reached. 420 421 If num_results is specified and the account has more than that number of 422 queues, the generator will have a populated next_marker field once it 423 finishes. This marker can be used to create a new generator if more 424 results are desired. 425 426 :param str prefix: 427 Filters the results to return only queues with names that begin 428 with the specified prefix. 429 :param int num_results: 430 The maximum number of queues to return. 431 :param bool include_metadata: 432 Specifies that container metadata be returned in the response. 433 :param str marker: 434 An opaque continuation token. This value can be retrieved from the 435 next_marker field of a previous generator object if num_results was 436 specified and that generator has finished enumerating results. If 437 specified, this generator will begin returning results from the point 438 where the previous generator stopped. 439 :param int timeout: 440 The server timeout, expressed in seconds. This function may make multiple 441 calls to the service in which case the timeout value specified will be 442 applied to each individual call. 443 ''' 444 include = 'metadata' if include_metadata else None 445 operation_context = _OperationContext(location_lock=True) 446 kwargs = {'prefix': prefix, 'max_results': num_results, 'include': include, 447 'marker': marker, 'timeout': timeout, '_context': operation_context} 448 resp = self._list_queues(**kwargs) 449 450 return ListGenerator(resp, self._list_queues, (), kwargs) 451 452 def _list_queues(self, prefix=None, marker=None, max_results=None, 453 include=None, timeout=None, _context=None): 454 ''' 455 Returns a list of queues under the specified account. Makes a single list 456 request to the service. Used internally by the list_queues method. 457 458 :param str prefix: 459 Filters the results to return only queues with names that begin 460 with the specified prefix. 461 :param str marker: 462 A token which identifies the portion of the query to be 463 returned with the next query operation. The operation returns a 464 next_marker element within the response body if the list returned 465 was not complete. This value may then be used as a query parameter 466 in a subsequent call to request the next portion of the list of 467 queues. The marker value is opaque to the client. 468 :param int max_results: 469 The maximum number of queues to return. A single list request may 470 return up to 1000 queues and potentially a continuation token which 471 should be followed to get additional resutls. 472 :param str include: 473 Include this parameter to specify that the container's 474 metadata be returned as part of the response body. 475 :param int timeout: 476 The server timeout, expressed in seconds. 477 ''' 478 request = HTTPRequest() 479 request.method = 'GET' 480 request.host_locations = self._get_host_locations(secondary=True) 481 request.path = _get_path() 482 request.query = { 483 'comp': 'list', 484 'prefix': _to_str(prefix), 485 'marker': _to_str(marker), 486 'maxresults': _int_to_str(max_results), 487 'include': _to_str(include), 488 'timeout': _int_to_str(timeout) 489 } 490 491 return self._perform_request(request, _convert_xml_to_queues, operation_context=_context) 492 493 def create_queue(self, queue_name, metadata=None, fail_on_exist=False, timeout=None): 494 ''' 495 Creates a queue under the given account. 496 497 :param str queue_name: 498 The name of the queue to create. A queue name must be from 3 through 499 63 characters long and may only contain lowercase letters, numbers, 500 and the dash (-) character. The first and last letters in the queue 501 must be alphanumeric. The dash (-) character cannot be the first or 502 last character. Consecutive dash characters are not permitted in the 503 queue name. 504 :param metadata: 505 A dict containing name-value pairs to associate with the queue as 506 metadata. Note that metadata names preserve the case with which they 507 were created, but are case-insensitive when set or read. 508 :type metadata: dict(str, str) 509 :param bool fail_on_exist: 510 Specifies whether to throw an exception if the queue already exists. 511 :param int timeout: 512 The server timeout, expressed in seconds. 513 :return: 514 A boolean indicating whether the queue was created. If fail_on_exist 515 was set to True, this will throw instead of returning false. 516 :rtype: bool 517 ''' 518 _validate_not_none('queue_name', queue_name) 519 request = HTTPRequest() 520 request.method = 'PUT' 521 request.host_locations = self._get_host_locations() 522 request.path = _get_path(queue_name) 523 request.query = {'timeout': _int_to_str(timeout)} 524 _add_metadata_headers(metadata, request) 525 526 def _return_request(request): 527 return request 528 529 if not fail_on_exist: 530 try: 531 response = self._perform_request(request, parser=_return_request, 532 expected_errors=[_QUEUE_ALREADY_EXISTS_ERROR_CODE]) 533 if response.status == _HTTP_RESPONSE_NO_CONTENT: 534 return False 535 return True 536 except AzureHttpError as ex: 537 _dont_fail_on_exist(ex) 538 return False 539 else: 540 response = self._perform_request(request, parser=_return_request) 541 if response.status == _HTTP_RESPONSE_NO_CONTENT: 542 raise AzureConflictHttpError( 543 _ERROR_CONFLICT.format(response.message), response.status) 544 return True 545 546 def delete_queue(self, queue_name, fail_not_exist=False, timeout=None): 547 ''' 548 Deletes the specified queue and any messages it contains. 549 550 When a queue is successfully deleted, it is immediately marked for deletion 551 and is no longer accessible to clients. The queue is later removed from 552 the Queue service during garbage collection. 553 554 Note that deleting a queue is likely to take at least 40 seconds to complete. 555 If an operation is attempted against the queue while it was being deleted, 556 an :class:`AzureConflictHttpError` will be thrown. 557 558 :param str queue_name: 559 The name of the queue to delete. 560 :param bool fail_not_exist: 561 Specifies whether to throw an exception if the queue doesn't exist. 562 :param int timeout: 563 The server timeout, expressed in seconds. 564 :return: 565 A boolean indicating whether the queue was deleted. If fail_not_exist 566 was set to True, this will throw instead of returning false. 567 :rtype: bool 568 ''' 569 _validate_not_none('queue_name', queue_name) 570 request = HTTPRequest() 571 request.method = 'DELETE' 572 request.host_locations = self._get_host_locations() 573 request.path = _get_path(queue_name) 574 request.query = {'timeout': _int_to_str(timeout)} 575 if not fail_not_exist: 576 try: 577 self._perform_request(request, expected_errors=[_QUEUE_NOT_FOUND_ERROR_CODE]) 578 return True 579 except AzureHttpError as ex: 580 _dont_fail_not_exist(ex) 581 return False 582 else: 583 self._perform_request(request) 584 return True 585 586 def get_queue_metadata(self, queue_name, timeout=None): 587 ''' 588 Retrieves user-defined metadata and queue properties on the specified 589 queue. Metadata is associated with the queue as name-value pairs. 590 591 :param str queue_name: 592 The name of an existing queue. 593 :param int timeout: 594 The server timeout, expressed in seconds. 595 :return: 596 A dictionary representing the queue metadata with an 597 approximate_message_count int property on the dict estimating the 598 number of messages in the queue. 599 :rtype: dict(str, str) 600 ''' 601 _validate_not_none('queue_name', queue_name) 602 request = HTTPRequest() 603 request.method = 'GET' 604 request.host_locations = self._get_host_locations(secondary=True) 605 request.path = _get_path(queue_name) 606 request.query = { 607 'comp': 'metadata', 608 'timeout': _int_to_str(timeout), 609 } 610 611 return self._perform_request(request, _parse_metadata_and_message_count) 612 613 def set_queue_metadata(self, queue_name, metadata=None, timeout=None): 614 ''' 615 Sets user-defined metadata on the specified queue. Metadata is 616 associated with the queue as name-value pairs. 617 618 :param str queue_name: 619 The name of an existing queue. 620 :param dict metadata: 621 A dict containing name-value pairs to associate with the 622 queue as metadata. 623 :param int timeout: 624 The server timeout, expressed in seconds. 625 ''' 626 _validate_not_none('queue_name', queue_name) 627 request = HTTPRequest() 628 request.method = 'PUT' 629 request.host_locations = self._get_host_locations() 630 request.path = _get_path(queue_name) 631 request.query = { 632 'comp': 'metadata', 633 'timeout': _int_to_str(timeout), 634 } 635 _add_metadata_headers(metadata, request) 636 637 self._perform_request(request) 638 639 def exists(self, queue_name, timeout=None): 640 ''' 641 Returns a boolean indicating whether the queue exists. 642 643 :param str queue_name: 644 The name of queue to check for existence. 645 :param int timeout: 646 The server timeout, expressed in seconds. 647 :return: A boolean indicating whether the queue exists. 648 :rtype: bool 649 ''' 650 _validate_not_none('queue_name', queue_name) 651 652 try: 653 request = HTTPRequest() 654 request.method = 'GET' 655 request.host_locations = self._get_host_locations(secondary=True) 656 request.path = _get_path(queue_name) 657 request.query = { 658 'comp': 'metadata', 659 'timeout': _int_to_str(timeout), 660 } 661 662 self._perform_request(request, expected_errors=[_QUEUE_NOT_FOUND_ERROR_CODE]) 663 return True 664 except AzureHttpError as ex: 665 _dont_fail_not_exist(ex) 666 return False 667 668 def get_queue_acl(self, queue_name, timeout=None): 669 ''' 670 Returns details about any stored access policies specified on the 671 queue that may be used with Shared Access Signatures. 672 673 :param str queue_name: 674 The name of an existing queue. 675 :param int timeout: 676 The server timeout, expressed in seconds. 677 :return: A dictionary of access policies associated with the queue. 678 :rtype: dict(str, :class:`~..common.models.AccessPolicy`) 679 ''' 680 _validate_not_none('queue_name', queue_name) 681 request = HTTPRequest() 682 request.method = 'GET' 683 request.host_locations = self._get_host_locations(secondary=True) 684 request.path = _get_path(queue_name) 685 request.query = { 686 'comp': 'acl', 687 'timeout': _int_to_str(timeout), 688 } 689 690 return self._perform_request(request, _convert_xml_to_signed_identifiers) 691 692 def set_queue_acl(self, queue_name, signed_identifiers=None, timeout=None): 693 ''' 694 Sets stored access policies for the queue that may be used with Shared 695 Access Signatures. 696 697 When you set permissions for a queue, the existing permissions are replaced. 698 To update the queue's permissions, call :func:`~get_queue_acl` to fetch 699 all access policies associated with the queue, modify the access policy 700 that you wish to change, and then call this function with the complete 701 set of data to perform the update. 702 703 When you establish a stored access policy on a queue, it may take up to 704 30 seconds to take effect. During this interval, a shared access signature 705 that is associated with the stored access policy will throw an 706 :class:`AzureHttpError` until the access policy becomes active. 707 708 :param str queue_name: 709 The name of an existing queue. 710 :param signed_identifiers: 711 A dictionary of access policies to associate with the queue. The 712 dictionary may contain up to 5 elements. An empty dictionary 713 will clear the access policies set on the service. 714 :type signed_identifiers: dict(str, :class:`~..common.models.AccessPolicy`) 715 :param int timeout: 716 The server timeout, expressed in seconds. 717 ''' 718 _validate_not_none('queue_name', queue_name) 719 _validate_access_policies(signed_identifiers) 720 request = HTTPRequest() 721 request.method = 'PUT' 722 request.host_locations = self._get_host_locations() 723 request.path = _get_path(queue_name) 724 request.query = { 725 'comp': 'acl', 726 'timeout': _int_to_str(timeout), 727 } 728 request.body = _get_request_body( 729 _convert_signed_identifiers_to_xml(signed_identifiers)) 730 self._perform_request(request) 731 732 def put_message(self, queue_name, content, visibility_timeout=None, 733 time_to_live=None, timeout=None): 734 ''' 735 Adds a new message to the back of the message queue. 736 737 The visibility timeout specifies the time that the message will be 738 invisible. After the timeout expires, the message will become visible. 739 If a visibility timeout is not specified, the default value of 0 is used. 740 741 The message time-to-live specifies how long a message will remain in the 742 queue. The message will be deleted from the queue when the time-to-live 743 period expires. 744 745 If the key-encryption-key field is set on the local service object, this method will 746 encrypt the content before uploading. 747 748 :param str queue_name: 749 The name of the queue to put the message into. 750 :param obj content: 751 Message content. Allowed type is determined by the encode_function 752 set on the service. Default is str. The encoded message can be up to 753 64KB in size. 754 :param int visibility_timeout: 755 If not specified, the default value is 0. Specifies the 756 new visibility timeout value, in seconds, relative to server time. 757 The value must be larger than or equal to 0, and cannot be 758 larger than 7 days. The visibility timeout of a message cannot be 759 set to a value later than the expiry time. visibility_timeout 760 should be set to a value smaller than the time-to-live value. 761 :param int time_to_live: 762 Specifies the time-to-live interval for the message, in 763 seconds. The time-to-live may be any positive number or -1 for infinity. If this 764 parameter is omitted, the default time-to-live is 7 days. 765 :param int timeout: 766 The server timeout, expressed in seconds. 767 :return: 768 A :class:`~azure.storage.queue.models.QueueMessage` object. 769 This object is also populated with the content although it is not 770 returned from the service. 771 :rtype: :class:`~azure.storage.queue.models.QueueMessage` 772 ''' 773 774 _validate_encryption_required(self.require_encryption, self.key_encryption_key) 775 776 _validate_not_none('queue_name', queue_name) 777 _validate_not_none('content', content) 778 request = HTTPRequest() 779 request.method = 'POST' 780 request.host_locations = self._get_host_locations() 781 request.path = _get_path(queue_name, True) 782 request.query = { 783 'visibilitytimeout': _to_str(visibility_timeout), 784 'messagettl': _to_str(time_to_live), 785 'timeout': _int_to_str(timeout) 786 } 787 788 request.body = _get_request_body(_convert_queue_message_xml(content, self.encode_function, 789 self.key_encryption_key)) 790 791 message_list = self._perform_request(request, _convert_xml_to_queue_messages, 792 [self.decode_function, False, 793 None, None, content]) 794 return message_list[0] 795 796 def get_messages(self, queue_name, num_messages=None, 797 visibility_timeout=None, timeout=None): 798 ''' 799 Retrieves one or more messages from the front of the queue. 800 801 When a message is retrieved from the queue, the response includes the message 802 content and a pop_receipt value, which is required to delete the message. 803 The message is not automatically deleted from the queue, but after it has 804 been retrieved, it is not visible to other clients for the time interval 805 specified by the visibility_timeout parameter. 806 807 If the key-encryption-key or resolver field is set on the local service object, the messages will be 808 decrypted before being returned. 809 810 :param str queue_name: 811 The name of the queue to get messages from. 812 :param int num_messages: 813 A nonzero integer value that specifies the number of 814 messages to retrieve from the queue, up to a maximum of 32. If 815 fewer are visible, the visible messages are returned. By default, 816 a single message is retrieved from the queue with this operation. 817 :param int visibility_timeout: 818 Specifies the new visibility timeout value, in seconds, relative 819 to server time. The new value must be larger than or equal to 1 820 second, and cannot be larger than 7 days. The visibility timeout of 821 a message can be set to a value later than the expiry time. 822 :param int timeout: 823 The server timeout, expressed in seconds. 824 :return: A :class:`~azure.storage.queue.models.QueueMessage` object representing the information passed. 825 :rtype: list(:class:`~azure.storage.queue.models.QueueMessage`) 826 ''' 827 _validate_decryption_required(self.require_encryption, self.key_encryption_key, 828 self.key_resolver_function) 829 830 _validate_not_none('queue_name', queue_name) 831 request = HTTPRequest() 832 request.method = 'GET' 833 request.host_locations = self._get_host_locations() 834 request.path = _get_path(queue_name, True) 835 request.query = { 836 'numofmessages': _to_str(num_messages), 837 'visibilitytimeout': _to_str(visibility_timeout), 838 'timeout': _int_to_str(timeout) 839 } 840 841 return self._perform_request(request, _convert_xml_to_queue_messages, 842 [self.decode_function, self.require_encryption, 843 self.key_encryption_key, self.key_resolver_function]) 844 845 def peek_messages(self, queue_name, num_messages=None, timeout=None): 846 ''' 847 Retrieves one or more messages from the front of the queue, but does 848 not alter the visibility of the message. 849 850 Only messages that are visible may be retrieved. When a message is retrieved 851 for the first time with a call to get_messages, its dequeue_count property 852 is set to 1. If it is not deleted and is subsequently retrieved again, the 853 dequeue_count property is incremented. The client may use this value to 854 determine how many times a message has been retrieved. Note that a call 855 to peek_messages does not increment the value of DequeueCount, but returns 856 this value for the client to read. 857 858 If the key-encryption-key or resolver field is set on the local service object, the messages will be 859 decrypted before being returned. 860 861 :param str queue_name: 862 The name of the queue to peek messages from. 863 :param int num_messages: 864 A nonzero integer value that specifies the number of 865 messages to peek from the queue, up to a maximum of 32. By default, 866 a single message is peeked from the queue with this operation. 867 :param int timeout: 868 The server timeout, expressed in seconds. 869 :return: 870 A list of :class:`~azure.storage.queue.models.QueueMessage` objects. Note that 871 time_next_visible and pop_receipt will not be populated as peek does 872 not pop the message and can only retrieve already visible messages. 873 :rtype: list(:class:`~azure.storage.queue.models.QueueMessage`) 874 ''' 875 876 _validate_decryption_required(self.require_encryption, self.key_encryption_key, 877 self.key_resolver_function) 878 879 _validate_not_none('queue_name', queue_name) 880 request = HTTPRequest() 881 request.method = 'GET' 882 request.host_locations = self._get_host_locations(secondary=True) 883 request.path = _get_path(queue_name, True) 884 request.query = { 885 'peekonly': 'true', 886 'numofmessages': _to_str(num_messages), 887 'timeout': _int_to_str(timeout) 888 } 889 890 return self._perform_request(request, _convert_xml_to_queue_messages, 891 [self.decode_function, self.require_encryption, 892 self.key_encryption_key, self.key_resolver_function]) 893 894 def delete_message(self, queue_name, message_id, pop_receipt, timeout=None): 895 ''' 896 Deletes the specified message. 897 898 Normally after a client retrieves a message with the get_messages operation, 899 the client is expected to process and delete the message. To delete the 900 message, you must have two items of data: id and pop_receipt. The 901 id is returned from the previous get_messages operation. The 902 pop_receipt is returned from the most recent :func:`~get_messages` or 903 :func:`~update_message` operation. In order for the delete_message operation 904 to succeed, the pop_receipt specified on the request must match the 905 pop_receipt returned from the :func:`~get_messages` or :func:`~update_message` 906 operation. 907 908 :param str queue_name: 909 The name of the queue from which to delete the message. 910 :param str message_id: 911 The message id identifying the message to delete. 912 :param str pop_receipt: 913 A valid pop receipt value returned from an earlier call 914 to the :func:`~get_messages` or :func:`~update_message`. 915 :param int timeout: 916 The server timeout, expressed in seconds. 917 ''' 918 _validate_not_none('queue_name', queue_name) 919 _validate_not_none('message_id', message_id) 920 _validate_not_none('pop_receipt', pop_receipt) 921 request = HTTPRequest() 922 request.method = 'DELETE' 923 request.host_locations = self._get_host_locations() 924 request.path = _get_path(queue_name, True, message_id) 925 request.query = { 926 'popreceipt': _to_str(pop_receipt), 927 'timeout': _int_to_str(timeout) 928 } 929 self._perform_request(request) 930 931 def clear_messages(self, queue_name, timeout=None): 932 ''' 933 Deletes all messages from the specified queue. 934 935 :param str queue_name: 936 The name of the queue whose messages to clear. 937 :param int timeout: 938 The server timeout, expressed in seconds. 939 ''' 940 _validate_not_none('queue_name', queue_name) 941 request = HTTPRequest() 942 request.method = 'DELETE' 943 request.host_locations = self._get_host_locations() 944 request.path = _get_path(queue_name, True) 945 request.query = {'timeout': _int_to_str(timeout)} 946 self._perform_request(request) 947 948 def update_message(self, queue_name, message_id, pop_receipt, visibility_timeout, 949 content=None, timeout=None): 950 ''' 951 Updates the visibility timeout of a message. You can also use this 952 operation to update the contents of a message. 953 954 This operation can be used to continually extend the invisibility of a 955 queue message. This functionality can be useful if you want a worker role 956 to "lease" a queue message. For example, if a worker role calls get_messages 957 and recognizes that it needs more time to process a message, it can 958 continually extend the message's invisibility until it is processed. If 959 the worker role were to fail during processing, eventually the message 960 would become visible again and another worker role could process it. 961 962 If the key-encryption-key field is set on the local service object, this method will 963 encrypt the content before uploading. 964 965 :param str queue_name: 966 The name of the queue containing the message to update. 967 :param str message_id: 968 The message id identifying the message to update. 969 :param str pop_receipt: 970 A valid pop receipt value returned from an earlier call 971 to the :func:`~get_messages` or :func:`~update_message` operation. 972 :param int visibility_timeout: 973 Specifies the new visibility timeout value, in seconds, 974 relative to server time. The new value must be larger than or equal 975 to 0, and cannot be larger than 7 days. The visibility timeout of a 976 message cannot be set to a value later than the expiry time. A 977 message can be updated until it has been deleted or has expired. 978 :param obj content: 979 Message content. Allowed type is determined by the encode_function 980 set on the service. Default is str. 981 :param int timeout: 982 The server timeout, expressed in seconds. 983 :return: 984 A list of :class:`~azure.storage.queue.models.QueueMessage` objects. For convenience, 985 this object is also populated with the content, although it is not returned by the service. 986 :rtype: list(:class:`~azure.storage.queue.models.QueueMessage`) 987 ''' 988 989 _validate_encryption_required(self.require_encryption, self.key_encryption_key) 990 991 _validate_not_none('queue_name', queue_name) 992 _validate_not_none('message_id', message_id) 993 _validate_not_none('pop_receipt', pop_receipt) 994 _validate_not_none('visibility_timeout', visibility_timeout) 995 request = HTTPRequest() 996 request.method = 'PUT' 997 request.host_locations = self._get_host_locations() 998 request.path = _get_path(queue_name, True, message_id) 999 request.query = { 1000 'popreceipt': _to_str(pop_receipt), 1001 'visibilitytimeout': _int_to_str(visibility_timeout), 1002 'timeout': _int_to_str(timeout) 1003 } 1004 1005 if content is not None: 1006 request.body = _get_request_body(_convert_queue_message_xml(content, self.encode_function, 1007 self.key_encryption_key)) 1008 1009 return self._perform_request(request, _parse_queue_message_from_headers) 1010