OLD | NEW |
(Empty) | |
| 1 # Copyright (c) 2006-2009 Mitch Garnaat http://garnaat.org/ |
| 2 # |
| 3 # Permission is hereby granted, free of charge, to any person obtaining a |
| 4 # copy of this software and associated documentation files (the |
| 5 # "Software"), to deal in the Software without restriction, including |
| 6 # without limitation the rights to use, copy, modify, merge, publish, dis- |
| 7 # tribute, sublicense, and/or sell copies of the Software, and to permit |
| 8 # persons to whom the Software is furnished to do so, subject to the fol- |
| 9 # lowing conditions: |
| 10 # |
| 11 # The above copyright notice and this permission notice shall be included |
| 12 # in all copies or substantial portions of the Software. |
| 13 # |
| 14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
| 15 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
| 16 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
| 17 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
| 18 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
| 19 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
| 20 # IN THE SOFTWARE. |
| 21 |
| 22 from boto.connection import AWSQueryConnection |
| 23 from boto.sqs.regioninfo import SQSRegionInfo |
| 24 from boto.sqs.queue import Queue |
| 25 from boto.sqs.message import Message |
| 26 from boto.sqs.attributes import Attributes |
| 27 from boto.sqs.batchresults import BatchResults |
| 28 from boto.exception import SQSError, BotoServerError |
| 29 |
| 30 |
| 31 class SQSConnection(AWSQueryConnection): |
| 32 """ |
| 33 A Connection to the SQS Service. |
| 34 """ |
| 35 DefaultRegionName = 'us-east-1' |
| 36 DefaultRegionEndpoint = 'queue.amazonaws.com' |
| 37 APIVersion = '2012-11-05' |
| 38 DefaultContentType = 'text/plain' |
| 39 ResponseError = SQSError |
| 40 AuthServiceName = 'sqs' |
| 41 |
| 42 def __init__(self, aws_access_key_id=None, aws_secret_access_key=None, |
| 43 is_secure=True, port=None, proxy=None, proxy_port=None, |
| 44 proxy_user=None, proxy_pass=None, debug=0, |
| 45 https_connection_factory=None, region=None, path='/', |
| 46 security_token=None, validate_certs=True): |
| 47 if not region: |
| 48 region = SQSRegionInfo(self, self.DefaultRegionName, |
| 49 self.DefaultRegionEndpoint) |
| 50 self.region = region |
| 51 AWSQueryConnection.__init__(self, aws_access_key_id, |
| 52 aws_secret_access_key, |
| 53 is_secure, port, |
| 54 proxy, proxy_port, |
| 55 proxy_user, proxy_pass, |
| 56 self.region.endpoint, debug, |
| 57 https_connection_factory, path, |
| 58 security_token=security_token, |
| 59 validate_certs=validate_certs) |
| 60 self.auth_region_name = self.region.name |
| 61 |
| 62 def _required_auth_capability(self): |
| 63 return ['hmac-v4'] |
| 64 |
| 65 def create_queue(self, queue_name, visibility_timeout=None): |
| 66 """ |
| 67 Create an SQS Queue. |
| 68 |
| 69 :type queue_name: str or unicode |
| 70 :param queue_name: The name of the new queue. Names are |
| 71 scoped to an account and need to be unique within that |
| 72 account. Calling this method on an existing queue name |
| 73 will not return an error from SQS unless the value for |
| 74 visibility_timeout is different than the value of the |
| 75 existing queue of that name. This is still an expensive |
| 76 operation, though, and not the preferred way to check for |
| 77 the existence of a queue. See the |
| 78 :func:`boto.sqs.connection.SQSConnection.lookup` method. |
| 79 |
| 80 :type visibility_timeout: int |
| 81 :param visibility_timeout: The default visibility timeout for |
| 82 all messages written in the queue. This can be overridden |
| 83 on a per-message. |
| 84 |
| 85 :rtype: :class:`boto.sqs.queue.Queue` |
| 86 :return: The newly created queue. |
| 87 |
| 88 """ |
| 89 params = {'QueueName': queue_name} |
| 90 if visibility_timeout: |
| 91 params['Attribute.1.Name'] = 'VisibilityTimeout' |
| 92 params['Attribute.1.Value'] = int(visibility_timeout) |
| 93 return self.get_object('CreateQueue', params, Queue) |
| 94 |
| 95 def delete_queue(self, queue, force_deletion=False): |
| 96 """ |
| 97 Delete an SQS Queue. |
| 98 |
| 99 :type queue: A Queue object |
| 100 :param queue: The SQS queue to be deleted |
| 101 |
| 102 :type force_deletion: Boolean |
| 103 :param force_deletion: Normally, SQS will not delete a queue |
| 104 that contains messages. However, if the force_deletion |
| 105 argument is True, the queue will be deleted regardless of |
| 106 whether there are messages in the queue or not. USE WITH |
| 107 CAUTION. This will delete all messages in the queue as |
| 108 well. |
| 109 |
| 110 :rtype: bool |
| 111 :return: True if the command succeeded, False otherwise |
| 112 """ |
| 113 return self.get_status('DeleteQueue', None, queue.id) |
| 114 |
| 115 def get_queue_attributes(self, queue, attribute='All'): |
| 116 """ |
| 117 Gets one or all attributes of a Queue |
| 118 |
| 119 :type queue: A Queue object |
| 120 :param queue: The SQS queue to be deleted |
| 121 |
| 122 :type attribute: str |
| 123 :type attribute: The specific attribute requested. If not |
| 124 supplied, the default is to return all attributes. Valid |
| 125 attributes are: |
| 126 |
| 127 * ApproximateNumberOfMessages |
| 128 * ApproximateNumberOfMessagesNotVisible |
| 129 * VisibilityTimeout |
| 130 * CreatedTimestamp |
| 131 * LastModifiedTimestamp |
| 132 * Policy |
| 133 * ReceiveMessageWaitTimeSeconds |
| 134 |
| 135 :rtype: :class:`boto.sqs.attributes.Attributes` |
| 136 :return: An Attributes object containing request value(s). |
| 137 """ |
| 138 params = {'AttributeName' : attribute} |
| 139 return self.get_object('GetQueueAttributes', params, |
| 140 Attributes, queue.id) |
| 141 |
| 142 def set_queue_attribute(self, queue, attribute, value): |
| 143 params = {'Attribute.Name' : attribute, 'Attribute.Value' : value} |
| 144 return self.get_status('SetQueueAttributes', params, queue.id) |
| 145 |
| 146 def receive_message(self, queue, number_messages=1, |
| 147 visibility_timeout=None, attributes=None, |
| 148 wait_time_seconds=None): |
| 149 """ |
| 150 Read messages from an SQS Queue. |
| 151 |
| 152 :type queue: A Queue object |
| 153 :param queue: The Queue from which messages are read. |
| 154 |
| 155 :type number_messages: int |
| 156 :param number_messages: The maximum number of messages to read |
| 157 (default=1) |
| 158 |
| 159 :type visibility_timeout: int |
| 160 :param visibility_timeout: The number of seconds the message should |
| 161 remain invisible to other queue readers |
| 162 (default=None which uses the Queues default) |
| 163 |
| 164 :type attributes: str |
| 165 :param attributes: The name of additional attribute to return |
| 166 with response or All if you want all attributes. The |
| 167 default is to return no additional attributes. Valid |
| 168 values: |
| 169 * All |
| 170 * SenderId |
| 171 * SentTimestamp |
| 172 * ApproximateReceiveCount |
| 173 * ApproximateFirstReceiveTimestamp |
| 174 |
| 175 :type wait_time_seconds: int |
| 176 :param wait_time_seconds: The duration (in seconds) for which the call |
| 177 will wait for a message to arrive in the queue before returning. |
| 178 If a message is available, the call will return sooner than |
| 179 wait_time_seconds. |
| 180 |
| 181 :rtype: list |
| 182 :return: A list of :class:`boto.sqs.message.Message` objects. |
| 183 |
| 184 """ |
| 185 params = {'MaxNumberOfMessages' : number_messages} |
| 186 if visibility_timeout is not None: |
| 187 params['VisibilityTimeout'] = visibility_timeout |
| 188 if attributes is not None: |
| 189 self.build_list_params(params, attributes, 'AttributeName') |
| 190 if wait_time_seconds is not None: |
| 191 params['WaitTimeSeconds'] = wait_time_seconds |
| 192 return self.get_list('ReceiveMessage', params, |
| 193 [('Message', queue.message_class)], |
| 194 queue.id, queue) |
| 195 |
| 196 def delete_message(self, queue, message): |
| 197 """ |
| 198 Delete a message from a queue. |
| 199 |
| 200 :type queue: A :class:`boto.sqs.queue.Queue` object |
| 201 :param queue: The Queue from which messages are read. |
| 202 |
| 203 :type message: A :class:`boto.sqs.message.Message` object |
| 204 :param message: The Message to be deleted |
| 205 |
| 206 :rtype: bool |
| 207 :return: True if successful, False otherwise. |
| 208 """ |
| 209 params = {'ReceiptHandle' : message.receipt_handle} |
| 210 return self.get_status('DeleteMessage', params, queue.id) |
| 211 |
| 212 def delete_message_batch(self, queue, messages): |
| 213 """ |
| 214 Deletes a list of messages from a queue in a single request. |
| 215 |
| 216 :type queue: A :class:`boto.sqs.queue.Queue` object. |
| 217 :param queue: The Queue to which the messages will be written. |
| 218 |
| 219 :type messages: List of :class:`boto.sqs.message.Message` objects. |
| 220 :param messages: A list of message objects. |
| 221 """ |
| 222 params = {} |
| 223 for i, msg in enumerate(messages): |
| 224 prefix = 'DeleteMessageBatchRequestEntry' |
| 225 p_name = '%s.%i.Id' % (prefix, (i+1)) |
| 226 params[p_name] = msg.id |
| 227 p_name = '%s.%i.ReceiptHandle' % (prefix, (i+1)) |
| 228 params[p_name] = msg.receipt_handle |
| 229 return self.get_object('DeleteMessageBatch', params, BatchResults, |
| 230 queue.id, verb='POST') |
| 231 |
| 232 def delete_message_from_handle(self, queue, receipt_handle): |
| 233 """ |
| 234 Delete a message from a queue, given a receipt handle. |
| 235 |
| 236 :type queue: A :class:`boto.sqs.queue.Queue` object |
| 237 :param queue: The Queue from which messages are read. |
| 238 |
| 239 :type receipt_handle: str |
| 240 :param receipt_handle: The receipt handle for the message |
| 241 |
| 242 :rtype: bool |
| 243 :return: True if successful, False otherwise. |
| 244 """ |
| 245 params = {'ReceiptHandle' : receipt_handle} |
| 246 return self.get_status('DeleteMessage', params, queue.id) |
| 247 |
| 248 def send_message(self, queue, message_content, delay_seconds=None): |
| 249 params = {'MessageBody' : message_content} |
| 250 if delay_seconds: |
| 251 params['DelaySeconds'] = int(delay_seconds) |
| 252 return self.get_object('SendMessage', params, Message, |
| 253 queue.id, verb='POST') |
| 254 |
| 255 def send_message_batch(self, queue, messages): |
| 256 """ |
| 257 Delivers up to 10 messages to a queue in a single request. |
| 258 |
| 259 :type queue: A :class:`boto.sqs.queue.Queue` object. |
| 260 :param queue: The Queue to which the messages will be written. |
| 261 |
| 262 :type messages: List of lists. |
| 263 :param messages: A list of lists or tuples. Each inner |
| 264 tuple represents a single message to be written |
| 265 and consists of and ID (string) that must be unique |
| 266 within the list of messages, the message body itself |
| 267 which can be a maximum of 64K in length, and an |
| 268 integer which represents the delay time (in seconds) |
| 269 for the message (0-900) before the message will |
| 270 be delivered to the queue. |
| 271 """ |
| 272 params = {} |
| 273 for i, msg in enumerate(messages): |
| 274 p_name = 'SendMessageBatchRequestEntry.%i.Id' % (i+1) |
| 275 params[p_name] = msg[0] |
| 276 p_name = 'SendMessageBatchRequestEntry.%i.MessageBody' % (i+1) |
| 277 params[p_name] = msg[1] |
| 278 p_name = 'SendMessageBatchRequestEntry.%i.DelaySeconds' % (i+1) |
| 279 params[p_name] = msg[2] |
| 280 return self.get_object('SendMessageBatch', params, BatchResults, |
| 281 queue.id, verb='POST') |
| 282 |
| 283 def change_message_visibility(self, queue, receipt_handle, |
| 284 visibility_timeout): |
| 285 """ |
| 286 Extends the read lock timeout for the specified message from |
| 287 the specified queue to the specified value. |
| 288 |
| 289 :type queue: A :class:`boto.sqs.queue.Queue` object |
| 290 :param queue: The Queue from which messages are read. |
| 291 |
| 292 :type receipt_handle: str |
| 293 :param queue: The receipt handle associated with the message whose |
| 294 visibility timeout will be changed. |
| 295 |
| 296 :type visibility_timeout: int |
| 297 :param visibility_timeout: The new value of the message's visibility |
| 298 timeout in seconds. |
| 299 """ |
| 300 params = {'ReceiptHandle' : receipt_handle, |
| 301 'VisibilityTimeout' : visibility_timeout} |
| 302 return self.get_status('ChangeMessageVisibility', params, queue.id) |
| 303 |
| 304 def change_message_visibility_batch(self, queue, messages): |
| 305 """ |
| 306 A batch version of change_message_visibility that can act |
| 307 on up to 10 messages at a time. |
| 308 |
| 309 :type queue: A :class:`boto.sqs.queue.Queue` object. |
| 310 :param queue: The Queue to which the messages will be written. |
| 311 |
| 312 :type messages: List of tuples. |
| 313 :param messages: A list of tuples where each tuple consists |
| 314 of a :class:`boto.sqs.message.Message` object and an integer |
| 315 that represents the new visibility timeout for that message. |
| 316 """ |
| 317 params = {} |
| 318 for i, t in enumerate(messages): |
| 319 prefix = 'ChangeMessageVisibilityBatchRequestEntry' |
| 320 p_name = '%s.%i.Id' % (prefix, (i+1)) |
| 321 params[p_name] = t[0].id |
| 322 p_name = '%s.%i.ReceiptHandle' % (prefix, (i+1)) |
| 323 params[p_name] = t[0].receipt_handle |
| 324 p_name = '%s.%i.VisibilityTimeout' % (prefix, (i+1)) |
| 325 params[p_name] = t[1] |
| 326 return self.get_object('ChangeMessageVisibilityBatch', |
| 327 params, BatchResults, |
| 328 queue.id, verb='POST') |
| 329 |
| 330 def get_all_queues(self, prefix=''): |
| 331 """ |
| 332 Retrieves all queues. |
| 333 |
| 334 :keyword str prefix: Optionally, only return queues that start with |
| 335 this value. |
| 336 :rtype: list |
| 337 :returns: A list of :py:class:`boto.sqs.queue.Queue` instances. |
| 338 """ |
| 339 params = {} |
| 340 if prefix: |
| 341 params['QueueNamePrefix'] = prefix |
| 342 return self.get_list('ListQueues', params, [('QueueUrl', Queue)]) |
| 343 |
| 344 def get_queue(self, queue_name): |
| 345 """ |
| 346 Retrieves the queue with the given name, or ``None`` if no match |
| 347 was found. |
| 348 |
| 349 :param str queue_name: The name of the queue to retrieve. |
| 350 :rtype: :py:class:`boto.sqs.queue.Queue` or ``None`` |
| 351 :returns: The requested queue, or ``None`` if no match was found. |
| 352 """ |
| 353 params = {'QueueName': queue_name} |
| 354 try: |
| 355 return self.get_object('GetQueueUrl', params, Queue) |
| 356 except SQSError: |
| 357 return None |
| 358 |
| 359 lookup = get_queue |
| 360 |
| 361 # |
| 362 # Permissions methods |
| 363 # |
| 364 |
| 365 def add_permission(self, queue, label, aws_account_id, action_name): |
| 366 """ |
| 367 Add a permission to a queue. |
| 368 |
| 369 :type queue: :class:`boto.sqs.queue.Queue` |
| 370 :param queue: The queue object |
| 371 |
| 372 :type label: str or unicode |
| 373 :param label: A unique identification of the permission you are setting. |
| 374 Maximum of 80 characters ``[0-9a-zA-Z_-]`` |
| 375 Example, AliceSendMessage |
| 376 |
| 377 :type aws_account_id: str or unicode |
| 378 :param principal_id: The AWS account number of the principal |
| 379 who will be given permission. The principal must have an |
| 380 AWS account, but does not need to be signed up for Amazon |
| 381 SQS. For information about locating the AWS account |
| 382 identification. |
| 383 |
| 384 :type action_name: str or unicode |
| 385 :param action_name: The action. Valid choices are: |
| 386 * * |
| 387 * SendMessage |
| 388 * ReceiveMessage |
| 389 * DeleteMessage |
| 390 * ChangeMessageVisibility |
| 391 * GetQueueAttributes |
| 392 |
| 393 :rtype: bool |
| 394 :return: True if successful, False otherwise. |
| 395 |
| 396 """ |
| 397 params = {'Label': label, |
| 398 'AWSAccountId' : aws_account_id, |
| 399 'ActionName' : action_name} |
| 400 return self.get_status('AddPermission', params, queue.id) |
| 401 |
| 402 def remove_permission(self, queue, label): |
| 403 """ |
| 404 Remove a permission from a queue. |
| 405 |
| 406 :type queue: :class:`boto.sqs.queue.Queue` |
| 407 :param queue: The queue object |
| 408 |
| 409 :type label: str or unicode |
| 410 :param label: The unique label associated with the permission |
| 411 being removed. |
| 412 |
| 413 :rtype: bool |
| 414 :return: True if successful, False otherwise. |
| 415 """ |
| 416 params = {'Label': label} |
| 417 return self.get_status('RemovePermission', params, queue.id) |
OLD | NEW |