"""Amazon SQS transport module for Kombu. This package implements an AMQP-like interface on top of Amazons SQS service, with the goal of being optimized for high performance and reliability. The default settings for this module are focused now on high performance in task queue situations where tasks are small, idempotent and run very fast. SQS Features supported by this transport ======================================== Long Polling ------------ https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html Long polling is enabled by setting the ``wait_time_seconds`` transport option to a value between 1 and 20 (valid range: 0–20, where 0 disables long polling). Amazon supports up to 20 seconds. The default is 10 seconds. When ``wait_time_seconds`` is greater than 0, each ``ReceiveMessage`` API call to AWS will block on the *server side* for up to that many seconds, returning early only when a message arrives. This reduces empty responses and lowers SQS request costs compared to short polling (``wait_time_seconds=0``). Polling Interval ---------------- The ``polling_interval`` transport option (default: 1 second) controls the *client-side* delay added between consecutive polls when a poll returns no messages. After a successful (non-empty) poll the next poll is scheduled immediately; after an empty poll the transport waits ``polling_interval`` seconds before trying again. **Important — additive timing when both options are set:** ``wait_time_seconds`` and ``polling_interval`` are applied at different layers and their delays compose additively. For an empty poll the total time before the next ``ReceiveMessage`` call is issued is:: effective interval = wait_time_seconds + polling_interval For example, with the defaults (``wait_time_seconds=10``, ``polling_interval=1``) an empty queue is polled at most once every ~11 seconds: the AWS call blocks for up to 10 s, then the client waits an additional 1 s before rescheduling. To make ``polling_interval`` the *sole* timing control (e.g. for short polling), set ``wait_time_seconds=0``: .. code-block:: python app.conf.broker_transport_options = { 'wait_time_seconds': 0, # disable server-side long polling 'polling_interval': 5, # client waits 5 s after each empty poll } To rely entirely on long polling and remove the extra client-side delay, set ``polling_interval=0``: .. code-block:: python app.conf.broker_transport_options = { 'wait_time_seconds': 10, # server holds connection up to 10 s 'polling_interval': 0, # no additional client-side wait } Batch API Actions ----------------- https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-batch-api.html The default behavior of the SQS Channel.drain_events() method is to request up to the 'prefetch_count' messages on every request to SQS. These messages are stored locally in a deque object and passed back to the Transport until the deque is empty, before triggering a new API call to Amazon. This behavior dramatically speeds up the rate that you can pull tasks from SQS when you have short-running tasks (or a large number of workers). When a Celery worker has multiple queues to monitor, it will pull down up to 'prefetch_count' messages from queueA and work on them all before moving on to queueB. If queueB is empty, it will wait up until 'polling_interval' expires (plus any server-side long-polling delay from 'wait_time_seconds') before moving back and checking on queueA. Message Attributes ----------------- https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html SQS supports sending message attributes along with the message body. To use this feature, you can pass a 'message_attributes' as keyword argument to `basic_publish` method. Other Features supported by this transport ========================================== Predefined Queues ----------------- The default behavior of this transport is to use a single AWS credential pair in order to manage all SQS queues (e.g. listing queues, creating queues, polling queues, deleting messages). If it is preferable for your environment to use multiple AWS credentials, you can use the 'predefined_queues' setting inside the 'transport_options' map. This setting allows you to specify the SQS queue URL and AWS credentials for each of your queues. For example, if you have two queues which both already exist in AWS) you can tell this transport about them as follows: .. code-block:: python transport_options = { 'predefined_queues': { 'queue-1': { 'url': 'https://sqs.us-east-1.amazonaws.com/xxx/aaa', 'access_key_id': 'a', 'secret_access_key': 'b', 'backoff_policy': {1: 10, 2: 20, 3: 40, 4: 80, 5: 320, 6: 640}, # optional 'backoff_tasks': ['svc.tasks.tasks.task1'] # optional }, 'queue-2.fifo': { 'url': 'https://sqs.us-east-1.amazonaws.com/xxx/bbb.fifo', 'access_key_id': 'c', 'secret_access_key': 'd', 'backoff_policy': {1: 10, 2: 20, 3: 40, 4: 80, 5: 320, 6: 640}, # optional 'backoff_tasks': ['svc.tasks.tasks.task2'] # optional }, } 'sts_role_arn': 'arn:aws:iam:::role/STSTest', # optional 'sts_token_timeout': 900, # optional 'sts_token_buffer_time': 0, # optional, added in 5.6.0 } Note that FIFO and standard queues must be named accordingly (the name of a FIFO queue must end with the .fifo suffix). backoff_policy & backoff_tasks are optional arguments. These arguments automatically change the message visibility timeout, in order to have different times between specific task retries. This would apply after task failure. A message rejected without requeue is deleted from SQS, as it would otherwise be redelivered once its visibility timeout expires. It is left in the queue in these cases: * Its task is listed in backoff_tasks, so the backoff policy can retry it. * The queue has a redrive policy. SQS redelivers the message until it has been received maxReceiveCount times, and then moves it to the dead-letter queue. The policy is read with GetQueueAttributes and the answer is reused for 5 minutes, so a dead-letter queue added to a live queue applies within that time without reconnecting. If that call fails, the message is left in the queue. * DeleteMessage fails, for example because the receipt handle has expired. The error is logged, and SQS redelivers the message. AWS STS authentication is supported, by using sts_role_arn, and sts_token_timeout. sts_role_arn is the assumed IAM role ARN we are trying to access with. sts_token_timeout is the token timeout, defaults (and minimum) to 900 seconds. After the mentioned period, a new token will be created. -------------------- Predefined Exchanges -------------------- When using a fanout exchange with this transport, messages are sent to an AWS SNS, which then forwards the messages to all subscribed queues. The default behavior of this transport is to create the SNS topic when the exchange is first declared. However, it is also possible to use a predefined SNS topic instead of letting the transport create it. .. code-block:: python transport_options = { 'predefined_exchanges': { 'exchange-1': { 'arn': 'arn:aws:sns:us-east-1:xxx:exchange-1', 'access_key_id': 'a', 'secret_access_key': 'b', }, 'exchange-2.fifo': { 'arn': 'arn:aws:sns:us-east-1:xxx:exchange-2', 'access_key_id': 'c', 'secret_access_key': 'd', }, } } .. versionadded:: 5.6.0 sts_token_buffer_time (seconds) is the time by which you want to refresh your token earlier than its actual expiration time, defaults to 0 (no time buffer will be added), should be less than sts_token_timeout. If you authenticate using Okta_ (e.g. calling |gac|_), you can also specify a 'session_token' to connect to a queue. Note that those tokens have a limited lifetime and are therefore only suited for short-lived tests. .. _Okta: https://www.okta.com/ .. _gac: https://github.com/Nike-Inc/gimme-aws-creds#readme .. |gac| replace:: ``gimme-aws-creds`` Client config ------------- In some cases you may need to override the botocore config. You can do it as follows: .. code-block:: python transport_option = { 'client-config': { 'connect_timeout': 5, }, } For a complete list of settings you can adjust using this option see https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html When a consumer is cancelled, for example on a Celery warm shutdown, a ``ReceiveMessage`` call may still be in flight. Any messages it returns are made visible again with ``ChangeMessageVisibility`` so other consumers can receive them right away. This call blocks the shutdown, so it uses its own client with a 5 second connect and read timeout and no retries, applied on top of ``client-config``. On a degraded network each call then gives up after seconds rather than minutes, and the first failure stops the release for that queue. The messages left become visible again once their visibility timeout expires. .. versionchanged:: 5.7.0 Messages received after a consumer is cancelled are released back to the queue with short timeouts and no retries, instead of raising ``KeyError`` and staying invisible until their visibility timeout. Features ======== * Type: Virtual * Supports Direct: Yes * Supports Topic: Yes * Supports Fanout: Yes * Supports Priority: No * Supports TTL: No """ from __future__ import annotations import base64 import binascii import re import socket import string import uuid from datetime import datetime, timedelta, timezone from json import JSONDecodeError from queue import Empty from time import monotonic from typing import Any, Literal from botocore.client import BaseClient, Config from botocore.exceptions import ClientError from vine import ensure_promise, promise, transform from kombu.asynchronous import get_event_loop from kombu.asynchronous.aws.ext import boto3 from kombu.asynchronous.aws.ext import exceptions as aws_exceptions from kombu.asynchronous.aws.sqs.connection import AsyncSQSConnection from kombu.asynchronous.aws.sqs.message import AsyncMessage from kombu.log import get_logger from kombu.transport import virtual from kombu.transport.SQS.exceptions import (AccessDeniedQueueException, DoesNotExistQueueException, InvalidQueueException, UndefinedQueueException) from kombu.transport.SQS.SNS import SNS as SnsFanout from kombu.utils import scheduling from kombu.utils.encoding import bytes_to_str, safe_str from kombu.utils.json import dumps, loads from kombu.utils.objects import cached_property logger = get_logger(__name__) # dots are replaced by dash, dash remains dash, all other punctuation # replaced by underscore. CHARS_REPLACE_TABLE = { ord(c): 0x5f for c in string.punctuation if c not in '-_.' } CHARS_REPLACE_TABLE[0x2e] = 0x2d # '.' -> '-' #: SQS bulk get supports a maximum of 10 messages at a time. SQS_MAX_MESSAGES = 10 #: SQS accepts at most 900 seconds of per-message delay. MAX_DELAY_SECONDS = 900 _SUPPORTED_BOTO_SERVICES = Literal["sqs", "sns"] def maybe_int(x): """Try to convert x' to int, or return x' if that fails.""" try: return int(x) except (TypeError, ValueError): return x class QoS(virtual.QoS): """Quality of Service guarantees implementation for SQS.""" def reject(self, delivery_tag, requeue=False): super().reject(delivery_tag, requeue=requeue) routing_key, message, backoff_tasks, backoff_policy = \ self._extract_backoff_policy_configuration_and_message( delivery_tag) if routing_key and message and backoff_tasks and backoff_policy: self.apply_backoff_policy( routing_key, delivery_tag, backoff_policy, backoff_tasks) def _extract_backoff_policy_configuration_and_message(self, delivery_tag): try: message = self._delivered[delivery_tag] routing_key = message.delivery_info['routing_key'] except KeyError: return None, None, None, None if not routing_key or not message: return None, None, None, None queue_config = self.channel.predefined_queues.get(routing_key, {}) backoff_tasks = queue_config.get('backoff_tasks') backoff_policy = queue_config.get('backoff_policy') return routing_key, message, backoff_tasks, backoff_policy def is_backoff_task(self, delivery_tag): """Return True if a backoff policy retries this delivery's task.""" routing_key, message, backoff_tasks, backoff_policy = \ self._extract_backoff_policy_configuration_and_message( delivery_tag) if not (routing_key and message and backoff_tasks and backoff_policy): return False return (message.headers or {}).get('task') in backoff_tasks def apply_backoff_policy(self, routing_key, delivery_tag, backoff_policy, backoff_tasks): queue_url = self.channel._queue_cache[routing_key] task_name, number_of_retries = \ self.extract_task_name_and_number_of_retries(delivery_tag) if not task_name or not number_of_retries: return None policy_value = backoff_policy.get(number_of_retries) if task_name in backoff_tasks and policy_value is not None: c = self.channel.sqs(routing_key) c.change_message_visibility( QueueUrl=queue_url, ReceiptHandle=delivery_tag, VisibilityTimeout=policy_value ) def extract_task_name_and_number_of_retries(self, delivery_tag): message = self._delivered[delivery_tag] message_headers = message.headers task_name = message_headers['task'] number_of_retries = int( message.properties['delivery_info']['sqs_message'] ['Attributes']['ApproximateReceiveCount']) return task_name, number_of_retries class Channel(virtual.Channel): """SQS Channel.""" default_region = 'us-east-1' default_visibility_timeout = 1800 # 30 minutes. default_wait_time_seconds = 10 # up to 20 seconds max # Not client-config alone: botocore's default 60 s timeouts and retries # could hold up the shutdown for minutes on a degraded network. release_client_config = Config( connect_timeout=5, read_timeout=5, retries={'total_max_attempts': 1}, ) domain_format = 'kombu%(vhost)s' #: Seconds a queue's redrive policy is reused before it is read again. #: Deleting a rejected message can't be undone, so a dead-letter queue #: added while workers run must be noticed without a reconnect. redrive_policy_cache_ttl = 300 _asynsqs = None _sqs = None _fanout = None QoS = QoS # https://stackoverflow.com/questions/475074/regex-to-parse-or-validate-base64-data B64_REGEX = re.compile( rb'^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$') def __init__(self, *args, **kwargs): if boto3 is None: raise ImportError('boto3 is not installed') super().__init__(*args, **kwargs) self._validate_predifined_queues() # SQS blows up if you try to create a new queue when one already # exists but with a different visibility_timeout. This prepopulates # the queue_cache to protect us from recreating # queues that are known to already exist. self._update_queue_cache(self.queue_name_prefix) self.hub = kwargs.get('hub') or get_event_loop() def _validate_predifined_queues(self): """Check that standard and FIFO queues are named properly. AWS requires FIFO queues to have a name that ends with the .fifo suffix. """ for queue_name, q in self.predefined_queues.items(): fifo_url = q['url'].endswith('.fifo') fifo_name = queue_name.endswith('.fifo') if fifo_url and not fifo_name: raise InvalidQueueException( "Queue with url '{}' must have a name " "ending with .fifo".format(q['url']) ) elif not fifo_url and fifo_name: raise InvalidQueueException( "Queue with name '{}' is not a FIFO queue: " "'{}'".format(queue_name, q['url']) ) def _update_queue_cache(self, queue_name_prefix): if self.predefined_queues: for queue_name, q in self.predefined_queues.items(): self._queue_cache[queue_name] = q['url'] return resp = self.sqs().list_queues(QueueNamePrefix=queue_name_prefix) for url in resp.get('QueueUrls', []): queue_name = url.split('/')[-1] self._queue_cache[queue_name] = url def basic_consume(self, queue, no_ack, callback, consumer_tag, *args, **kwargs): # If using a Fanout exchange, then subscribe to the queue to SNS self._subscribe_queue_to_fanout_exchange_if_required(queue) if no_ack: self._noack_queues.add(queue) if self.hub: # Start the polling loop immediately on first consume without # incurring any polling_interval delay at startup. self.hub.call_soon(self._schedule_queue, queue) return super().basic_consume(queue, no_ack, callback, consumer_tag, *args, **kwargs) def basic_cancel(self, consumer_tag): if consumer_tag in self._consumers: queue = self._tag_to_queue[consumer_tag] self._noack_queues.discard(queue) return super().basic_cancel(consumer_tag) def _queue_bind(self, exchange, routing_key, pattern, queue): # If the exchange is a fanout exchange, initialise the SNS topic if self._exchange_is_fanout(exchange): self.fanout.initialise_exchange(exchange) def drain_events(self, timeout=None, callback=None, **kwargs): """Return a single payload message from one of our queues. Raises ------ Queue.Empty: if no messages available. """ # If we're not allowed to consume or have no consumers, raise Empty if not self._consumers or not self.qos.can_consume(): raise Empty() # At this point, go and get more messages from SQS self._poll(self.cycle, callback, timeout=timeout) def _reset_cycle(self): """Reset the consume cycle. Returns ------- FairCycle: object that points to our _get_bulk() method rather than the standard _get() method. This allows for multiple messages to be returned at once from SQS ( based on the prefetch limit). """ self._cycle = scheduling.FairCycle( self._get_bulk, self._active_queues, Empty, ) def entity_name(self, name, table=CHARS_REPLACE_TABLE): """Format AMQP queue name into a legal SQS queue name.""" if name.endswith('.fifo'): partial = name[:-len('.fifo')] partial = str(safe_str(partial)).translate(table) return partial + '.fifo' else: return str(safe_str(name)).translate(table) def canonical_queue_name(self, queue_name): return self.entity_name(self.queue_name_prefix + queue_name) def _resolve_queue_url(self, queue): """Try to retrieve the SQS queue URL for a given queue name.""" # Translate to SQS name for consistency with initial # _queue_cache population. sqs_qname = self.canonical_queue_name(queue) # The SQS ListQueues method only returns 1000 queues. When you have # so many queues, it's possible that the queue you are looking for is # not cached. In this case, we could update the cache with the exact # queue name first. if sqs_qname not in self._queue_cache: self._update_queue_cache(sqs_qname) try: return self._queue_cache[sqs_qname] except KeyError: if self.predefined_queues: raise UndefinedQueueException(( "Queue with name '{}' must be " "defined in 'predefined_queues'." ).format(sqs_qname)) raise DoesNotExistQueueException( f"Queue with name '{sqs_qname}' doesn't exist in SQS" ) def _new_queue(self, queue, **kwargs): """Ensure a queue with given name exists in SQS. Arguments: --------- queue (str): the AMQP queue name Returns str: the SQS queue URL """ try: return self._resolve_queue_url(queue) except DoesNotExistQueueException: sqs_qname = self.canonical_queue_name(queue) attributes = {'VisibilityTimeout': str(self.visibility_timeout)} if sqs_qname.endswith('.fifo'): attributes['FifoQueue'] = 'true' resp = self._create_queue(sqs_qname, attributes) queue_url = self._queue_cache[sqs_qname] = resp["QueueUrl"] return queue_url def _create_queue(self, queue_name, attributes): """Create an SQS queue with a given name and nominal attributes.""" # Allow specifying additional boto create_queue Attributes # via transport options if self.predefined_queues: raise UndefinedQueueException( f"Queue with name '{queue_name}' must be defined in 'predefined_queues'." ) attributes.update( self.transport_options.get('sqs-creation-attributes') or {}, ) queue_tags = self.transport_options.get('queue_tags') create_params = { 'QueueName': queue_name, 'Attributes': attributes, } if queue_tags: create_params['tags'] = queue_tags return self.sqs(queue=queue_name).create_queue(**create_params) def _delete(self, queue, exchange: str | None = None, *args, **kwargs): """Delete queue by name. :param queue: The queue name :param exchange: The exchange name, if any :return: None """ if self.predefined_queues: return q_url = self._resolve_queue_url(queue) self.sqs().delete_queue( QueueUrl=q_url, ) # If the exchange is a fanout exchange, unsubscribe the queue to the SNS topic if exchange and self._exchange_is_fanout(exchange): self.fanout.subscriptions.unsubscribe_queue( queue_name=queue, exchange_name=exchange ) self._queue_cache.pop(queue, None) def _resolve_delay_seconds(self, properties): """Return the per-message delay to apply, or ``None`` for no delay. Accepts the delay either as ``delay_seconds`` or as ``DelaySeconds``, so callers can use the snake_case spelling that matches Kombu's own ``Producer.publish`` style. A value of ``None`` under one spelling means it was not supplied there, so the other spelling is consulted. Values are coerced to ``int`` (a float truncates toward zero) and clamped to the range SQS accepts, since anything above the maximum is rejected by AWS at publish time. Arguments: --------- properties (dict): The message properties to read the delay from. Returns ------- int: The delay in seconds -- including ``0``, which SQS accepts and treats as no delay -- or ``None`` when no delay was requested or the requested value cannot be honoured. """ for key in ('delay_seconds', 'DelaySeconds'): if key not in properties: continue value = properties[key] if value is None: # Not supplied under this spelling; the other one may carry it. continue try: delay = int(value) except (TypeError, ValueError, OverflowError): logger.warning( 'SQS delay %r is not a number, publishing without a delay', value, ) return None if delay < 0: logger.warning( 'SQS delay of %s is negative, publishing without a delay', delay, ) return None if delay > MAX_DELAY_SECONDS: logger.warning( 'SQS delay of %s exceeds the maximum of %s seconds, ' 'clamping to the maximum', delay, MAX_DELAY_SECONDS, ) delay = MAX_DELAY_SECONDS return delay return None def _put(self, queue, message, **kwargs): """Put message onto queue.""" q_url = self._new_queue(queue) kwargs = {'QueueUrl': q_url} if 'properties' in message: if 'message_attributes' in message['properties']: # we don't want to want to have the attribute in the body kwargs['MessageAttributes'] = \ message['properties'].pop('message_attributes') # Support SQS fair queue system. if 'MessageGroupId' in message['properties']: kwargs['MessageGroupId'] = \ message['properties']['MessageGroupId'] delay_seconds = self._resolve_delay_seconds( message['properties']) # Support FIFO queues. if queue.endswith('.fifo'): if 'MessageGroupId' not in kwargs: kwargs['MessageGroupId'] = 'default' if 'MessageDeduplicationId' in message['properties']: kwargs['MessageDeduplicationId'] = \ message['properties']['MessageDeduplicationId'] else: kwargs['MessageDeduplicationId'] = str(uuid.uuid4()) if delay_seconds is not None: # SQS rejects DelaySeconds on a FIFO queue outright, so it # is dropped here -- loudly, rather than silently. logger.warning( 'SQS FIFO queues take no per-message delay, ignoring ' 'delay of %s seconds', delay_seconds, ) elif delay_seconds is not None: kwargs['DelaySeconds'] = delay_seconds if self.sqs_base64_encoding: body = AsyncMessage().encode(dumps(message)) else: body = dumps(message) kwargs['MessageBody'] = body c = self.sqs(queue=self.canonical_queue_name(queue)) if message.get('redelivered'): c.change_message_visibility( QueueUrl=q_url, ReceiptHandle=message['properties']['delivery_tag'], VisibilityTimeout=self.wait_time_seconds ) else: c.send_message(**kwargs) def _message_to_python(self, message, queue_name, q_url): raw_msg_body = message['Body'] decoded_bytes = self._decode_python_message_body(raw_msg_body) text = bytes_to_str(decoded_bytes) payload = self._prepare_json_payload(text) # handle no-ack queues immediately if queue_name in self._noack_queues: self._delete_message(queue_name, message) return payload return self._envelope_payload(payload, text, message, q_url) def _messages_to_python(self, messages, queue): """Convert a list of SQS Message objects into Payloads. This method handles converting SQS Message objects into Payloads, and appropriately updating the queue depending on the 'ack' settings for that queue. Arguments: --------- messages (SQSMessage): A list of SQS Message objects. queue (str): Name representing the queue they came from. Returns ------- List: A list of Payload objects """ q_url = self._new_queue(queue) return [self._message_to_python(m, queue, q_url) for m in messages] def _receive_message( self, queue: str, max_number_of_messages: int = 1, wait_time_seconds: int | None = None ): """Unified receive_message wrapper for SQS (boto3.client.SQS) with full attribute support. :param queue: The queue as a string :param max_number_of_messages: Int of max number of messages to receive. :param wait_time_seconds: Int of sqs wait time in seconds. :return: SQS client recieve_message """ q_url: str = self._new_queue(queue) client = self.sqs(queue=queue) msg_attrs = self.get_message_attributes params: dict[str, Any] = { 'QueueUrl': q_url, 'MaxNumberOfMessages': max_number_of_messages, 'WaitTimeSeconds': wait_time_seconds or self.wait_time_seconds, } if msg_sys_attrs := msg_attrs.get('MessageSystemAttributeNames'): params['MessageSystemAttributeNames'] = msg_sys_attrs if msg_attrs := msg_attrs.get('MessageAttributeNames'): params['MessageAttributeNames'] = msg_attrs return client.receive_message(**params) def _get_bulk(self, queue, max_if_unlimited=SQS_MAX_MESSAGES, callback=None): """Try to retrieve multiple messages off ``queue``. Where :meth:`_get` returns a single Payload object, this method returns a list of Payload objects. The number of objects returned is determined by the total number of messages available in the queue and the number of messages the QoS object allows (based on the prefetch_count). Note: ---- Ignores QoS limits so caller is responsible for checking that we are allowed to consume at least one message from the queue. get_bulk will then ask QoS for an estimate of the number of extra messages that we can consume. Arguments: --------- queue (str): The queue name to pull from. Returns ------- List[Message] """ # drain_events calls `can_consume` first, consuming # a token, so we know that we are allowed to consume at least # one message. # Note: ignoring max_messages for SQS with boto3 max_count = self._get_message_estimate() if max_count: resp = self._receive_message( queue=queue, wait_time_seconds=self.wait_time_seconds, max_number_of_messages=max_count ) if messages := resp.get("Messages"): for m in messages: m["Body"] = AsyncMessage(body=m["Body"]).decode() for msg in self._messages_to_python(messages, queue): self.connection._deliver(msg, queue) return raise Empty() def _get(self, queue): """Try to retrieve a single message off ``queue``.""" resp = self._receive_message( queue=queue, wait_time_seconds=self.wait_time_seconds, max_number_of_messages=1 ) if messages := resp.get("Messages"): body = AsyncMessage(body=messages[0]["Body"]).decode() messages[0]["Body"] = body return self._messages_to_python(messages, queue)[0] raise Empty() def _loop1(self, queue, messages_fetched=None): if messages_fetched: # Messages were found in the last poll; schedule next poll immediately. self.hub.call_soon(self._schedule_queue, queue) else: # No messages found (unsuccessful poll); respect polling_interval # before the next attempt to avoid hammering SQS unnecessarily. # Note: self.connection is the Transport instance (virtual transport # establish_connection() returns self), so polling_interval is # accessed directly on self.connection. polling_interval = self.connection.polling_interval if polling_interval: self.hub.call_later(polling_interval, self._schedule_queue, queue) return self.hub.call_soon(self._schedule_queue, queue) def _schedule_queue(self, queue): if queue in self._active_queues: if self.qos.can_consume(): self._get_bulk_async( queue, callback=promise(self._loop1, (queue,)), ) else: self._loop1(queue) def _get_message_estimate(self, max_if_unlimited=SQS_MAX_MESSAGES): maxcount = self.qos.can_consume_max_estimate() return min( max_if_unlimited if maxcount is None else max(maxcount, 1), max_if_unlimited, ) def _get_bulk_async(self, queue, callback=None): maxcount = self._get_message_estimate() if maxcount: return self._get_async(queue, maxcount, callback=callback) # Not allowed to consume, make sure to notify callback.. callback = ensure_promise(callback) callback([]) return callback def _get_async(self, queue, count=1, callback=None): q_url = self._new_queue(queue) qname = self.canonical_queue_name(queue) return self._get_from_sqs( queue_name=qname, queue_url=q_url, count=count, connection=self.asynsqs(queue=qname), callback=transform( self._on_messages_ready, callback, q_url, queue ), ) def _on_messages_ready(self, queue, qname, messages): msgs = messages.get('Messages') if not msgs: return 0 callback = self.connection._callbacks.get(qname) if callback is None: # Not routed through Transport._deliver(): its requeue path # parses each message first (deleting it on no-ack queues) and # hides it again for wait_time_seconds, one call per message. self._release_messages(queue, qname, msgs) return 0 for msg in msgs: msg_parsed = self._message_to_python(msg, qname, queue) callback(msg_parsed) return len(msgs) def _release_messages(self, queue, qname, messages): """Make received messages visible again without consuming them.""" logger.info( 'Consumer for queue %s is gone, releasing %d received message(s)', qname, len(messages), ) # The blocking client is used instead of asynsqs(): the hub may stop # right after the consumer is cancelled, before an async call returns. # ChangeMessageVisibilityBatch is not used because GoAws, which the # integration tests run against, does not implement it. try: client = self._new_release_client( self.sqs(queue=self.canonical_queue_name(qname)), ) for msg in messages: client.change_message_visibility( QueueUrl=queue, ReceiptHandle=msg['ReceiptHandle'], VisibilityTimeout=0, ) except (ClientError, aws_exceptions.BotoCoreError, OSError) as exc: # The messages left become visible again once their visibility # timeout expires, so they are delayed but not lost. logger.warning( 'Failed to release messages for queue %s: %r', qname, exc, ) def _new_release_client(self, client): """Copy *client* with the short timeouts of release_client_config.""" # Not copied when new_sqs_client() is overridden: the copy would drop # whatever the override customises, such as the endpoint or hooks. if type(self).new_sqs_client is not Channel.new_sqs_client: return client # The credentials are taken from the client because sqs() does not # keep them, e.g. the STS credentials of a predefined queue. credentials = client._request_signer._credentials if credentials is None: return client frozen = credentials.get_frozen_credentials() return self._new_boto_client( service='sqs', region=client.meta.region_name, access_key_id=frozen.access_key, secret_access_key=frozen.secret_key, session_token=frozen.token, config=self.release_client_config, ) def _get_from_sqs(self, queue_name, queue_url, connection, count=1, callback=None): """Retrieve and handle messages from SQS. Uses long polling and returns :class:`~vine.promises.promise`. """ return connection.receive_message( queue_name, queue_url, number_messages=count, wait_time_seconds=self.wait_time_seconds, callback=callback, ) def _restore(self, message, unwanted_delivery_info=('sqs_message', 'sqs_queue')): for unwanted_key in unwanted_delivery_info: # Remove objects that aren't JSON serializable (Issue #1108). message.delivery_info.pop(unwanted_key, None) return super()._restore(message) def basic_ack(self, delivery_tag, multiple=False): try: message = self.qos.get(delivery_tag).delivery_info sqs_message = message['sqs_message'] except KeyError: super().basic_ack(delivery_tag) else: queue = None if 'routing_key' in message: queue = self.canonical_queue_name(message['routing_key']) try: self.sqs(queue=queue).delete_message( QueueUrl=message['sqs_queue'], ReceiptHandle=sqs_message['ReceiptHandle'] ) except ClientError as exception: error_code = exception.response['Error']['Code'] if error_code == 'AccessDenied': # Raised so the misconfiguration is not silently retried, # but the delivery is released first, as for any other # failed ack, so its prefetch slot is not held forever. super().basic_ack(delivery_tag) raise AccessDeniedQueueException( exception.response["Error"]["Message"] ) if error_code in ('InvalidParameterValue', 'ReceiptHandleIsInvalid'): # Forget the delivery without applying backoff to a stale handle. super().basic_ack(delivery_tag) else: super().basic_reject(delivery_tag) except BaseException: # DeleteMessage did not complete (it may or may not have # reached SQS: a read timeout can lose only the response). If # the message was not deleted, SQS redelivers it after the # visibility timeout. Either way the delivery must be released # here: otherwise its prefetch slot is held for the life of the # channel, and once every slot is held ``can_consume()`` stays # False and the consumer stops polling. Re-raise so the caller # still sees and logs the failure. super().basic_ack(delivery_tag) raise else: super().basic_ack(delivery_tag) def basic_reject(self, delivery_tag, requeue=False): try: if not requeue and not self.qos.is_backoff_task(delivery_tag): # A message rejected without requeue is discarded. Unless it # is deleted, SQS redelivers it once its visibility timeout # expires, so it would be received again and again forever. # Tasks with a backoff policy are left in the queue, as that # policy relies on SQS redelivering them. self._delete_rejected_message(delivery_tag) finally: super().basic_reject(delivery_tag, requeue=requeue) def _delete_rejected_message(self, delivery_tag): try: message = self.qos.get(delivery_tag).delivery_info sqs_message = message['sqs_message'] queue_url = message['sqs_queue'] except KeyError: return queue = None if 'routing_key' in message: queue = self.canonical_queue_name(message['routing_key']) if self._has_redrive_policy(queue, queue_url): # SQS moves the message to the dead-letter queue once it has been # received maxReceiveCount times, so it must not be deleted. return try: self.sqs(queue=queue).delete_message( QueueUrl=queue_url, ReceiptHandle=sqs_message['ReceiptHandle'] ) except ClientError as exception: # The message is not deleted, so SQS redelivers it once its # visibility timeout expires. Raising would stop the worker, as # the caller only handles the transport's connection errors. logger.warning( 'Could not delete rejected message from %s: %s', queue_url, exception, ) def _has_redrive_policy(self, queue, queue_url): now = monotonic() cached = self._redrive_policies.get(queue_url) if cached is not None: has_policy, read_at = cached if now - read_at < self.redrive_policy_cache_ttl: return has_policy try: resp = self.sqs(queue=queue).get_queue_attributes( QueueUrl=queue_url, AttributeNames=['RedrivePolicy'], ) except ClientError as exception: logger.warning( 'Could not read the redrive policy of %s, so rejected ' 'messages are left in the queue: %s', queue_url, exception, ) if exception.response['Error']['Code'] == 'AccessDenied': # Asking again right away would be denied again. self._redrive_policies[queue_url] = (True, now) return True has_policy = 'RedrivePolicy' in resp.get('Attributes', {}) self._redrive_policies[queue_url] = (has_policy, now) return has_policy def _size(self, queue): """Return the number of messages in a queue.""" q_url = self._new_queue(queue) c = self.sqs(queue=self.canonical_queue_name(queue)) resp = c.get_queue_attributes( QueueUrl=q_url, AttributeNames=['ApproximateNumberOfMessages'] ) return int(resp['Attributes']['ApproximateNumberOfMessages']) def _purge(self, queue): """Delete all current messages in a queue.""" q_url = self._new_queue(queue) # SQS is slow at registering messages, so run for a few # iterations to ensure messages are detected and deleted. size = 0 for i in range(10): size += int(self._size(queue)) if not size: break self.sqs(queue=queue).purge_queue(QueueUrl=q_url) return size def close(self): super().close() # if self._asynsqs: # try: # self.asynsqs().close() # except AttributeError as exc: # FIXME ??? # if "can't set attribute" not in str(exc): # raise def new_sqs_client( self, region, access_key_id, secret_access_key, session_token=None ): """Create a new SQS client. :param region: The AWS region to use. :param access_key_id: The AWS access key ID for authenticating with boto. :param secret_access_key: The AWS secret access key for authenticating with boto. :param session_token: The AWS session token for authenticating with boto, if required. :returns: A Boto SQS client. """ return self._new_boto_client( service="sqs", region=region, access_key_id=access_key_id, secret_access_key=secret_access_key, session_token=session_token, ) def _new_boto_client( self, service: _SUPPORTED_BOTO_SERVICES, region, access_key_id, secret_access_key, session_token=None, config=None, ): session = boto3.Session( region_name=region, aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, aws_session_token=session_token, ) is_secure = self.is_secure if self.is_secure is not None else True client_kwargs = { 'use_ssl': is_secure } if self.endpoint_url is not None: client_kwargs['endpoint_url'] = self.endpoint_url client_config = Config( **(self.transport_options.get('client-config') or {}), ) if config is not None: client_config = client_config.merge(config) return session.client(service, config=client_config, **client_kwargs) def sqs(self, queue=None): # If a queue has been provided, check if the queue has been defined already. Reuse it's client if possible. if queue is not None and self.predefined_queues: # Raise if queue is not defined if queue not in self.predefined_queues: raise UndefinedQueueException( f"Queue with name '{queue}' must be defined in 'predefined_queues'." ) q = self.predefined_queues[queue] # Handle authenticating boto client with tokens if self.transport_options.get("sts_role_arn"): return self._handle_sts_session(queue, q) # If the queue has already been defined, then return the client for the queue if c := self._predefined_queue_clients.get(queue): return c # Create client, add it to the queue map and return c = self._predefined_queue_clients[queue] = self.new_sqs_client( region=q.get("region", self.region), access_key_id=q.get("access_key_id", self.conninfo.userid), secret_access_key=q.get("secret_access_key", self.conninfo.password), ) return c # If SQS client has been initialised, return it if self._sqs is not None: return self._sqs # Initialise a new SQS client and return it c = self._sqs = self.new_sqs_client( region=self.region, access_key_id=self.conninfo.userid, secret_access_key=self.conninfo.password, ) return c @property def fanout(self) -> SnsFanout: """Provides SNS fanout functionality. This method returns the fanout instance. If an instance of the fanout class has not been initialised, then initialise it. :returns: An instance of SNS fanout class. """ # If an SNS class has not been initialised, then initialise it if not self._fanout: self._fanout = SnsFanout(self) return self._fanout def remove_stale_sns_subscriptions(self, exchange_name: str) -> None: """Removes any stale SNS topic subscriptions. This method will check that any SQS subscriptions on the SNS topic are associated with SQS queues. If not, it will remove the stale subscription. This method will only work if the 'supports_fanout' property is True. :param exchange_name: The exchange to check for stale subscriptions :return: None """ if self._exchange_is_fanout(exchange_name): return self.fanout.subscriptions.cleanup(exchange_name) return None def _handle_sts_session(self, queue: str, q): """Checks if the STS token needs renewing for SQS. :param queue: The queue name :param q: The queue object :returns: The SQS client with a refreshed STS token """ region = q.get("region", self.region) # Check if a token refresh is needed if self.is_sts_token_refresh_required( name=queue, client_map=self._predefined_queue_clients, expire_time=getattr(self, "sts_expiration", None), ): return self._new_predefined_queue_client_with_sts_session(queue, region) # If token refresh is not required, return existing client return self._predefined_queue_clients[queue] @staticmethod def is_sts_token_refresh_required( name: Any, client_map: dict[str, BaseClient], expire_time: datetime | None = None, ) -> bool: """Checks if the STS token needs renewing. This method will check different STS expiry times depending on the service the token was used for. :param name: Either the queue name or exchange name :param client_map: Map of client names to boto3 clients. Either the queue or exchange map :param expire_time: The datetime when the token expires. :returns: True if the token needs renewing, False otherwise. """ # Get the expiry time of the STS token depending on the service if not expire_time: # STS token - token init return True # STS token - refresh if expired elif ( expire_time.replace(tzinfo=None) < datetime.now(timezone.utc).replace(tzinfo=None) ): return True # STS token = refresh if exchange or queue is not in client map elif not client_map.get(name): return True # STS token - reuse existing else: return False def generate_sts_session_token_with_buffer( self, role_arn, token_expiry_seconds, token_buffer_seconds=0 ): """Generate STS session credentials with an optional expiration buffer. The buffer is only applied if it is less than `token_expiry_seconds` to prevent an expired token. """ credentials = self.generate_sts_session_token(role_arn, token_expiry_seconds) if token_buffer_seconds and 0 < token_buffer_seconds < token_expiry_seconds: credentials["Expiration"] -= timedelta(seconds=token_buffer_seconds) return credentials def _new_predefined_queue_client_with_sts_session(self, queue, region): # Handle STS token refresh sts_creds = self.get_sts_credentials() self.sts_expiration = sts_creds["Expiration"] # Get new client and return it c = self._predefined_queue_clients[queue] = self.new_sqs_client( region=region, access_key_id=sts_creds["AccessKeyId"], secret_access_key=sts_creds["SecretAccessKey"], session_token=sts_creds["SessionToken"], ) return c def get_sts_credentials(self): return self.generate_sts_session_token_with_buffer( self.transport_options.get("sts_role_arn"), self.transport_options.get("sts_token_timeout", 900), self.transport_options.get("sts_token_buffer_time", 0), ) @staticmethod def generate_sts_session_token(role_arn: str, token_expiry_seconds: int): sts_client = boto3.client("sts") sts_policy = sts_client.assume_role( RoleArn=role_arn, RoleSessionName='Celery', DurationSeconds=token_expiry_seconds ) return sts_policy['Credentials'] def asynsqs(self, queue=None): message_system_attribute_names = self.get_message_attributes.get( "MessageSystemAttributeNames" ) message_attribute_names = self.get_message_attributes.get( "MessageAttributeNames" ) if queue is not None and self.predefined_queues: if queue in self._predefined_queue_async_clients and not hasattr( self, "sts_expiration" ): return self._predefined_queue_async_clients[queue] if not (q := self.predefined_queues.get(queue)): raise UndefinedQueueException( f"Queue with name '{queue}' must be defined in 'predefined_queues'." ) c = self._predefined_queue_async_clients[queue] = AsyncSQSConnection( sqs_connection=self.sqs(queue=queue), region=q.get("region", self.region), message_system_attribute_names=message_system_attribute_names, message_attribute_names=message_attribute_names, ) return c if self._asynsqs is not None: return self._asynsqs c = self._asynsqs = AsyncSQSConnection( sqs_connection=self.sqs(queue=queue), region=self.region, message_system_attribute_names=message_system_attribute_names, message_attribute_names=message_attribute_names ) return c @property def conninfo(self): return self.connection.client @property def transport_options(self): return self.connection.client.transport_options @property def _queue_cache(self): return self.connection._queue_cache @property def _redrive_policies(self): return self.connection._redrive_policies @property def _noack_queues(self): return self.connection._noack_queues @property def _predefined_queue_clients(self): return self.connection._predefined_queue_clients @property def _predefined_queue_async_clients(self): return self.connection._predefined_queue_async_clients @cached_property def visibility_timeout(self): # sqs only accepts whole seconds, and 0 is a valid value, so the # default only applies when the option is absent or empty. timeout = self.transport_options.get('visibility_timeout') if timeout is None or timeout == '': timeout = self.default_visibility_timeout return int(float(timeout)) @cached_property def predefined_queues(self): """Map of queue_name to predefined queue settings.""" return self.transport_options.get("predefined_queues", {}) @cached_property def predefined_exchanges(self): """Map of exchange_name to predefined SNS client.""" return self.transport_options.get("predefined_exchanges", {}) @cached_property def queue_name_prefix(self): return self.transport_options.get('queue_name_prefix', '') @cached_property def supports_fanout(self): return self.transport_options.get("supports_fanout", False) @cached_property def region(self): return (self.transport_options.get('region') or boto3.Session().region_name or self.default_region) @cached_property def regioninfo(self): return self.transport_options.get('regioninfo') @cached_property def is_secure(self): return self.transport_options.get('is_secure') @cached_property def port(self): return self.transport_options.get('port') @cached_property def endpoint_url(self): if self.conninfo.hostname is not None: scheme = 'https' if self.is_secure else 'http' if self.conninfo.port is not None: port = f':{self.conninfo.port}' else: port = "" return f"{scheme}://{self.conninfo.hostname}{port}" @cached_property def wait_time_seconds(self) -> int: return self.transport_options.get( "wait_time_seconds", self.default_wait_time_seconds ) @cached_property def sqs_base64_encoding(self): return self.transport_options.get('sqs_base64_encoding', True) @cached_property def fetch_message_attributes(self): return self.transport_options.get('fetch_message_attributes', None) @property def get_message_attributes(self) -> dict[str, Any]: """Get the message attributes to be fetched from SQS. Ensures 'ApproximateReceiveCount' is included in system attributes if list is provided. - The number of retries is managed by SQS / (specifically by the ``ApproximateReceiveCount`` message attribute) - See: class QoS(virtual.QoS): (method) def extract_task_name_and_number_of_retries :return: A dictionary with SQS message attribute fetch config. """ APPROXIMATE_RECEIVE_COUNT = 'ApproximateReceiveCount' fetch = self.fetch_message_attributes message_system_attrs = None message_attrs = None if fetch is None or isinstance(fetch, str): return { 'MessageAttributeNames': [], 'MessageSystemAttributeNames': [APPROXIMATE_RECEIVE_COUNT], } if isinstance(fetch, list): message_system_attrs = ['ALL'] if 'ALL'.lower() in [s.lower() for s in fetch] else ( list(set(fetch + [APPROXIMATE_RECEIVE_COUNT])) ) elif isinstance(fetch, dict): system = fetch.get('MessageSystemAttributeNames', []) attrs = fetch.get('MessageAttributeNames', None) if isinstance(system, list): message_system_attrs = ['ALL'] if 'ALL'.lower() in [s.lower() for s in system] else ( list(set(system + [APPROXIMATE_RECEIVE_COUNT])) ) if isinstance(attrs, list) and attrs: message_attrs = ['ALL'] if 'ALL'.lower() in [s.lower() for s in attrs] else ( list(set(attrs)) ) return { "MessageAttributeNames": sorted(message_attrs) if message_attrs else [], "MessageSystemAttributeNames": ( sorted(message_system_attrs) if message_system_attrs else [APPROXIMATE_RECEIVE_COUNT] ), } def _put_fanout(self, exchange: str, message: dict, routing_key, **kwargs): """Add a message to fanout queues by adding a notification to an SNS topic, with subscribed SQS queues. :param exchange: The name of the exchange to add the notification to. :param message: The message to be added. :param routing_key: The routing key to use for the notification. :param kwargs: Additional parameters :return: None """ # Extract properties and message attributes from message for SNS parameters request_params = {} properties = message.get("properties", {}) message_attrs = properties.get("message_attributes") # If the exchange is a FIFO topic, then add required MessageGroupId and # MessageDeduplicationId attributes if exchange.endswith(".fifo"): request_params["MessageGroupId"] = properties.get( "MessageGroupId", "default" ) request_params["MessageDeduplicationId"] = properties.get( "MessageDeduplicationId", str(uuid.uuid4()) ) # Add the message to the SNS topic self.fanout.publish( exchange_name=exchange, message=dumps(message), message_attributes=message_attrs, request_params=request_params, ) def _subscribe_queue_to_fanout_exchange_if_required(self, queue_name: str) -> None: """Subscribe the given queue to the SNS topic if this is a fanout exchange. :param queue_name: The name of the queue to subscribe. :return: None """ try: # Get the exchange name for the queue and get the exchange type exchange_name = self._get_exchange_for_queue(queue_name) # If the exchange is a fanout type and the transport supports it, # subscribe the queue to the topic if self._exchange_is_fanout(exchange_name): self.fanout.subscriptions.subscribe_queue( queue_name=queue_name, exchange_name=exchange_name ) except UndefinedQueueException as e: logger.debug( f"Not subscribing queue '{queue_name}' to fanout exchange: {e}" ) def _exchange_is_fanout(self, exchange_name: str) -> bool: """Check if the given exchange is a fanout type. :param exchange_name: The name of the exchange to check. :return: True if the exchange is a fanout type and the transport supports it, False otherwise. """ try: exchange_type = self.state.exchanges[exchange_name]["type"] return exchange_type == "fanout" and self.supports_fanout except KeyError: return False def _get_exchange_for_queue(self, queue_name: str) -> str: """Get the exchange name for the given queue. :param queue_name: The name of the queue to get the exchange for. :return: The name of the exchange for the given queue. :raises UndefinedQueueException: If the queue has not been defined. """ try: return list(self.state.queue_index[queue_name])[0].exchange except (KeyError, IndexError): raise UndefinedQueueException(f"Queue '{queue_name}' has not been defined.") # ————————————————————————————————————————————————————————————— # _message_to_python helper methods (extracted for testing/readability) # ————————————————————————————————————————————————————————————— def _optional_b64_decode(self, raw: bytes) -> bytes: """Optionally decode a base64 encoded string. :param raw: The raw bytes object to decode. :return: Bytes of the optionally decoded raw input. """ candidate = raw.strip() if self.B64_REGEX.fullmatch(candidate) is None: return raw try: decoded = base64.b64decode(candidate, validate=True) except (binascii.Error, ValueError): return raw reencoded = base64.b64encode(decoded).rstrip(b'=') if reencoded != candidate.rstrip(b'='): return raw try: decoded.decode('utf-8') except UnicodeDecodeError: return raw return decoded def _decode_python_message_body(self, raw_body): """Decode the message body when needed. raw_body: bytes or str returns: bytes (decoded Base64 if it looks like Base64, otherwise raw bytes) """ b = raw_body.encode() if isinstance(raw_body, str) else raw_body return self._optional_b64_decode(b) def _prepare_json_payload(self, text): """Try to JSON-decode text into a dict; on failure return {}.""" try: data = loads(text) return data if isinstance(data, dict) else {} except (JSONDecodeError, TypeError): return {} def _delete_message(self, queue_name, message): """Move the message over to the new queue URL and delete it.""" new_q = self._new_queue(queue_name) self.asynsqs(queue=queue_name).delete_message( new_q, message['ReceiptHandle'] ) def _envelope_payload(self, payload, raw_text, message, q_url): """Prepare the payload envelope. Ensure we have a dict with 'body' and 'properties.delivery_info', then stamp on SQS-specific metadata. :param payload: The payload as an object :param raw_text: Text that will be set as the payload body. :param message: A kombu Message. :param q_url: The SQS queue URL. :return: Payload object. """ # if payload wasn’t already a Kombu JSON dict, wrap it if 'properties' not in payload: payload = { 'body': raw_text, 'properties': {'delivery_info': {}}, } props = payload.setdefault('properties', {}) di = props.setdefault('delivery_info', {}) # add SQS metadata di.update({ 'sqs_message': message, 'sqs_queue': q_url, }) props['delivery_tag'] = message['ReceiptHandle'] return payload class Transport(virtual.Transport): """SQS Transport. Additional queue attributes can be supplied to SQS during queue creation by passing an ``sqs-creation-attributes`` key in transport_options. ``sqs-creation-attributes`` must be a dict whose key-value pairs correspond with Attributes in the `CreateQueue SQS API`_. For example, to have SQS queues created with server-side encryption enabled using the default Amazon Managed Customer Master Key, you can set ``KmsMasterKeyId`` Attribute. When the queue is initially created by Kombu, encryption will be enabled. .. code-block:: python from kombu.transport.SQS import Transport transport = Transport( ..., transport_options={ 'sqs-creation-attributes': { 'KmsMasterKeyId': 'alias/aws/sqs', }, } ) .. _CreateQueue SQS API: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_CreateQueue.html#API_CreateQueue_RequestParameters .. versionadded:: 5.6 Queue tags can be applied to SQS queues during creation by passing an ``queue_tags`` key in transport_options. ``queue_tags`` must be a dict of tag key-value pairs. .. code-block:: python from kombu.transport.SQS import Transport transport = Transport( ..., transport_options={ 'queue_tags': { 'Environment': 'production', 'Team': 'backend', }, } ) The ``ApproximateReceiveCount`` message attribute is fetched by this transport by default. Requested message attributes can be changed by setting ``fetch_message_attributes`` in the transport options. .. code-block:: python from kombu.transport.SQS import Transport transport = Transport( ..., transport_options={ 'fetch_message_attributes': ["All"], # Get all of the MessageSystemAttributeNames (formerly AttributeNames) } ) # Preferred - A dict specifying system and custom message attributes transport = Transport( ..., transport_options={ 'fetch_message_attributes': { 'MessageSystemAttributeNames': ["SenderId", "SentTimestamp"], 'MessageAttributeNames': ['S3MessageBodyKey'] }, } ) .. _Message Attributes: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html#SQS-ReceiveMessage-request-AttributeNames """ # noqa: E501 Channel = Channel polling_interval = 1 wait_time_seconds = 0 default_port = None connection_errors = ( virtual.Transport.connection_errors + (aws_exceptions.BotoCoreError, socket.error) ) channel_errors = ( virtual.Transport.channel_errors + (aws_exceptions.BotoCoreError,) ) driver_type = 'sqs' driver_name = 'sqs' implements = virtual.Transport.implements.extend( asynchronous=True, exchange_type=frozenset(['direct']), ) def __init__(self, client, **kwargs): super().__init__(client, **kwargs) # Scoped to this Connection: its channels share them, but another # Connection (different region, account or credentials) must not. self._queue_cache = {} # SQS queue name => SQS queue URL # SQS queue URL => (whether the queue has a redrive policy, # monotonic time it was read) self._redrive_policies = {} self._noack_queues = set() self._predefined_queue_clients = {} # A client for each predefined queue self._predefined_queue_async_clients = {} @property def default_connection_params(self): return {'port': self.default_port}