f86a2f6e-a611-48ec-a078-fed1bbc27bef

Committed 07 Feb 2025 02:46PM UTC coverage: 86.922% (-0.02%) from 86.943%

Build # f86a2f6e-a611-48ec-a078-fed1bbc27bef

Build Type

push

circleci

Committed by

web-flow

Commit Message

[ESM] Dynamically set SQS message override count (#12233)

Coverage Stats

11 of 11 new or added lines in 1 file covered. (100.0%)

21 existing lines in 12 files now uncovered.

61467 of 70715 relevant lines covered (86.92%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.33

/localstack-core/localstack/services/lambda_/event_source_mapping/pollers/sqs_poller.py

import json
import logging
from collections import defaultdict
from functools import cached_property

from botocore.client import BaseClient

from localstack.aws.api.pipes import PipeSourceSqsQueueParameters
from localstack.aws.api.sqs import MessageSystemAttributeName
from localstack.config import internal_service_url
from localstack.services.lambda_.event_source_mapping.event_processor import (
    EventProcessor,
    PartialBatchFailureError,
)
from localstack.services.lambda_.event_source_mapping.pollers.poller import (
    Poller,
    parse_batch_item_failures,
)
from localstack.services.sqs.constants import HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT
from localstack.utils.aws.arns import parse_arn
from localstack.utils.strings import first_char_to_lower

LOG = logging.getLogger(__name__)

DEFAULT_MAX_RECEIVE_COUNT = 10


class SqsPoller(Poller):
    queue_url: str

    batch_size: int
    maximum_batching_window: int

    def __init__(
        self,
        source_arn: str,
        source_parameters: dict | None = None,
        source_client: BaseClient | None = None,
        processor: EventProcessor | None = None,
    ):
        super().__init__(source_arn, source_parameters, source_client, processor)
        self.queue_url = get_queue_url(self.source_arn)

        self.batch_size = self.sqs_queue_parameters["BatchSize"]
        self.maximum_batching_window = self.sqs_queue_parameters["MaximumBatchingWindowInSeconds"]

        self._register_client_hooks()

    @property
    def sqs_queue_parameters(self) -> PipeSourceSqsQueueParameters:
        return self.source_parameters["SqsQueueParameters"]

    @cached_property
    def is_fifo_queue(self) -> bool:
        # Alternative heuristic: self.queue_url.endswith(".fifo"), but we need the call to get_queue_attributes for IAM
        return self.get_queue_attributes().get("FifoQueue", "false").lower() == "true"

    def _register_client_hooks(self):
        event_system = self.source_client.meta.events

        def handle_message_count_override(params, context, **kwargs):
            requested_count = params.pop("sqs_override_max_message_count", None)
            if not requested_count or requested_count <= DEFAULT_MAX_RECEIVE_COUNT:
                return

            context[HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = str(requested_count)

        def handle_inject_headers(params, context, **kwargs):
            if override := context.pop(HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT, None):
                params["headers"][HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = override

        event_system.register(
            "provide-client-params.sqs.ReceiveMessage", handle_message_count_override
        )
        # Since we delete SQS messages after processing, this allows us to remove up to 10K entries at a time.
        event_system.register(
            "provide-client-params.sqs.DeleteMessageBatch", handle_message_count_override
        )

        event_system.register("before-call.sqs.ReceiveMessage", handle_inject_headers)
        event_system.register("before-call.sqs.DeleteMessageBatch", handle_inject_headers)

    def get_queue_attributes(self) -> dict:
        """The API call to sqs:GetQueueAttributes is required for IAM policy streamsing."""
        get_queue_attributes_response = self.source_client.get_queue_attributes(
            QueueUrl=self.queue_url,
            AttributeNames=["FifoQueue"],
        )
        return get_queue_attributes_response.get("Attributes", {})

    def event_source(self) -> str:
        return "aws:sqs"

    def poll_events(self) -> None:
        # SQS pipe source: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
        # "The 9 Ways an SQS Message can be Deleted": https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/
        # TODO: implement batch window expires based on MaximumBatchingWindowInSeconds
        # TODO: implement invocation payload size quota
        # TODO: consider long-polling vs. short-polling trade-off. AWS uses long-polling:
        #  https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html#pipes-sqs-scaling
        response = self.source_client.receive_message(
            QueueUrl=self.queue_url,
            MaxNumberOfMessages=min(self.batch_size, DEFAULT_MAX_RECEIVE_COUNT),
            MessageAttributeNames=["All"],
            MessageSystemAttributeNames=[MessageSystemAttributeName.All],
            # Override how many messages we can receive per call
            sqs_override_max_message_count=self.batch_size,
        )
        if messages := response.get("Messages"):
            LOG.debug("Polled %d events from %s", len(messages), self.source_arn)
            try:
                if self.is_fifo_queue:
                    # TODO: think about starvation behavior because once failing message could block other groups
                    fifo_groups = split_by_message_group_id(messages)
                    for fifo_group_messages in fifo_groups.values():
                        self.handle_messages(fifo_group_messages)
                else:
                    self.handle_messages(messages)

            # TODO: unify exception handling across pollers: should we catch and raise?
            except Exception as e:
                # TODO: improve error messages (produce same failure and design better error messages)
                LOG.warning(
                    "Polling or batch processing failed: %s",
                    e,
                    exc_info=LOG.isEnabledFor(logging.DEBUG),
                )

    def handle_messages(self, messages):
        polled_events = transform_into_events(messages)
        # Filtering: matching vs. discarded (i.e., not matching filter criteria)
        # TODO: implement format detection behavior (e.g., for JSON body):
        #  https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
        #  Check whether we need poller-specific filter-preprocessing here without modifying the actual event!
        # convert to json for filtering (HACK for fixing parity with v1 and getting regression tests passing)
        for event in polled_events:
            try:
                event["body"] = json.loads(event["body"])
            except json.JSONDecodeError:
                LOG.debug(
                    "Unable to convert event body '%s' to json... Event might be dropped.",
                    event["body"],
                )
        matching_events = self.filter_events(polled_events)
        # convert them back (HACK for fixing parity with v1 and getting regression tests passing)
        for event in matching_events:
            event["body"] = (
                json.dumps(event["body"]) if not isinstance(event["body"], str) else event["body"]
            )

        all_message_ids = {message["MessageId"] for message in messages}
        matching_message_ids = {event["messageId"] for event in matching_events}
        discarded_message_ids = all_message_ids.difference(matching_message_ids)
        # Delete discarded events immediately:
        # https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/#7-event-source-mappings-with-filters
        self.delete_messages(messages, discarded_message_ids)

        # Don't trigger upon empty events
        if len(matching_events) == 0:
            return
        # Enrich events with metadata after filtering
        enriched_events = self.add_source_metadata(matching_events)

        # Invoke the processor (e.g., Pipe, ESM) and handle partial batch failures
        try:
            self.processor.process_events_batch(enriched_events)
            successful_message_ids = all_message_ids
        except PartialBatchFailureError as e:
            failed_message_ids = parse_batch_item_failures(
                e.partial_failure_payload, matching_message_ids
            )
            successful_message_ids = matching_message_ids.difference(failed_message_ids)

        # Only delete messages that are processed successfully as described here:
        # https://docs.aws.amazon.com/en_gb/lambda/latest/dg/with-sqs.html
        # When Lambda reads a batch, the messages stay in the queue but are hidden for the length of the queue's
        # visibility timeout. If your function successfully processes the batch, Lambda deletes the messages
        # from the queue. By default, if your function encounters an error while processing a batch,
        # all messages in that batch become visible in the queue again. For this reason, your function code must
        # be able to process the same message multiple times without unintended side effects.
        # Troubleshooting: https://repost.aws/knowledge-center/lambda-sqs-report-batch-item-failures
        # For FIFO queues, AWS also deletes successfully sent messages. Therefore, the AWS docs recommends:
        # "If you're using this feature with a FIFO queue, your function should stop processing messages after the first
        # failure and return all failed and unprocessed messages in batchItemFailures. This helps preserve the ordering
        # of messages in your queue."
        # Following this recommendation could result in the unsolved side effect that valid messages are continuously
        # placed in the same batch as failing messages:
        # * https://stackoverflow.com/questions/78694079/how-to-stop-fifo-sqs-messages-from-being-placed-in-a-batch-with-failing-messages
        # * https://stackoverflow.com/questions/76912394/can-i-report-only-messages-from-failing-group-id-in-reportbatchitemfailures-resp

        # TODO: Test blocking failure behavior for FIFO queues to guarantee strict ordering
        #  -> might require some checkpointing or retry control on the poller side?!
        # The poller should only proceed processing FIFO queues after having retried failing messages:
        # "If your pipe returns an error, the pipe attempts all retries on the affected messages before EventBridge
        # receives additional messages from the same group."
        # https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
        self.delete_messages(messages, successful_message_ids)

    def delete_messages(self, messages: list[dict], message_ids_to_delete: set):
        """Delete SQS `messages` from the source queue that match a MessageId within `message_ids_to_delete`"""
        # TODO: unclear how (partial) failures for deleting are handled, retry or fail batch? Hard to test against AWS
        if len(message_ids_to_delete) > 0:
            entries = [
                {"Id": str(count), "ReceiptHandle": message["ReceiptHandle"]}
                for count, message in enumerate(messages)
                if message["MessageId"] in message_ids_to_delete
            ]

            self.source_client.delete_message_batch(
                QueueUrl=self.queue_url,
                Entries=entries,
                # Override how many messages can be deleted at once
                sqs_override_max_message_count=self.batch_size,
            )


def split_by_message_group_id(messages) -> defaultdict[str, list[dict]]:
    """Splitting SQS messages by MessageGroupId to ensure strict ordering for FIFO queues"""
    fifo_groups = defaultdict(list)
    for message in messages:
        message_group_id = message["Attributes"]["MessageGroupId"]
        fifo_groups[message_group_id].append(message)
    return fifo_groups


def transform_into_events(messages: list[dict]) -> list[dict]:
    events = []
    for message in messages:
        # TODO: consolidate with SQS event source listener:
        #  localstack.services.lambda_.event_source_listeners.sqs_event_source_listener.SQSEventSourceListener._send_event_to_lambda
        message_attrs = message_attributes_to_lower(message.get("MessageAttributes"))
        event = {
            # Original SQS message attributes
            "messageId": message["MessageId"],
            "receiptHandle": message["ReceiptHandle"],
            # TODO: test with empty body
            # TODO: implement heuristic based on content type: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
            "body": message.get("Body", "MessageBody"),
            "attributes": message.get("Attributes", {}),
            "messageAttributes": message_attrs,
            # TODO: test with empty body
            "md5OfBody": message.get("MD5OfBody") or message.get("MD5OfMessageBody"),
        }
        # TODO: test Pipe with message attributes (only covered by Lambda ESM SQS test so far)
        if md5_of_message_attributes := message.get("MD5OfMessageAttributes"):
            event["md5OfMessageAttributes"] = md5_of_message_attributes
        events.append(event)
    return events


def get_queue_url(queue_arn: str) -> str:
    # TODO: consolidate this method with localstack.services.sqs.models.SqsQueue.url
    # * Do we need to support different endpoint strategies?
    # * If so, how can we achieve this without having a request context
    host_url = internal_service_url()
    host = host_url.rstrip("/")
    parsed_arn = parse_arn(queue_arn)
    account_id = parsed_arn["account"]
    name = parsed_arn["resource"]
    return f"{host}/{account_id}/{name}"


def message_attributes_to_lower(message_attrs):
    """Convert message attribute details (first characters) to lower case (e.g., stringValue, dataType)."""
    message_attrs = message_attrs or {}
    for _, attr in message_attrs.items():
        if not isinstance(attr, dict):
            continue
        for key, value in dict(attr).items():
            attr[first_char_to_lower(key)] = attr.pop(key)
    return message_attrs

1	import json	1✔
2	import logging	1✔
3	from collections import defaultdict	1✔
4	from functools import cached_property	1✔
5
6	from botocore.client import BaseClient	1✔
7
8	from localstack.aws.api.pipes import PipeSourceSqsQueueParameters	1✔
9	from localstack.aws.api.sqs import MessageSystemAttributeName	1✔
10	from localstack.config import internal_service_url	1✔
11	from localstack.services.lambda_.event_source_mapping.event_processor import (	1✔
12	EventProcessor,
13	PartialBatchFailureError,
14	)
15	from localstack.services.lambda_.event_source_mapping.pollers.poller import (	1✔
16	Poller,
17	parse_batch_item_failures,
18	)
19	from localstack.services.sqs.constants import HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT	1✔
20	from localstack.utils.aws.arns import parse_arn	1✔
21	from localstack.utils.strings import first_char_to_lower	1✔
22
23	LOG = logging.getLogger(__name__)	1✔
24
25	DEFAULT_MAX_RECEIVE_COUNT = 10	1✔
26
27
28	class SqsPoller(Poller):	1✔
29	queue_url: str	1✔
30
31	batch_size: int	1✔
32	maximum_batching_window: int	1✔
33
34	def __init__(	1✔
35	self,
36	source_arn: str,
37	source_parameters: dict \| None = None,
38	source_client: BaseClient \| None = None,
39	processor: EventProcessor \| None = None,
40	):
41	super().__init__(source_arn, source_parameters, source_client, processor)	1✔
42	self.queue_url = get_queue_url(self.source_arn)	1✔
43
44	self.batch_size = self.sqs_queue_parameters["BatchSize"]	1✔
45	self.maximum_batching_window = self.sqs_queue_parameters["MaximumBatchingWindowInSeconds"]	1✔
46
47	self._register_client_hooks()	1✔
48
49	@property	1✔
50	def sqs_queue_parameters(self) -> PipeSourceSqsQueueParameters:	1✔
51	return self.source_parameters["SqsQueueParameters"]	1✔
52
53	@cached_property	1✔
54	def is_fifo_queue(self) -> bool:	1✔
55	# Alternative heuristic: self.queue_url.endswith(".fifo"), but we need the call to get_queue_attributes for IAM
56	return self.get_queue_attributes().get("FifoQueue", "false").lower() == "true"	1✔
57
58	def _register_client_hooks(self):	1✔
59	event_system = self.source_client.meta.events	1✔
60
61	def handle_message_count_override(params, context, **kwargs):	1✔
62	requested_count = params.pop("sqs_override_max_message_count", None)	1✔
63	if not requested_count or requested_count <= DEFAULT_MAX_RECEIVE_COUNT:	1✔
64	return	1✔
65
66	context[HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = str(requested_count)	1✔
67
68	def handle_inject_headers(params, context, **kwargs):	1✔
69	if override := context.pop(HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT, None):	1✔
70	params["headers"][HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = override	1✔
71
72	event_system.register(	1✔
73	"provide-client-params.sqs.ReceiveMessage", handle_message_count_override
74	)
75	# Since we delete SQS messages after processing, this allows us to remove up to 10K entries at a time.
76	event_system.register(	1✔
77	"provide-client-params.sqs.DeleteMessageBatch", handle_message_count_override
78	)
79
80	event_system.register("before-call.sqs.ReceiveMessage", handle_inject_headers)	1✔
81	event_system.register("before-call.sqs.DeleteMessageBatch", handle_inject_headers)	1✔
82
83	def get_queue_attributes(self) -> dict:	1✔
84	"""The API call to sqs:GetQueueAttributes is required for IAM policy streamsing."""
85	get_queue_attributes_response = self.source_client.get_queue_attributes(	1✔
86	QueueUrl=self.queue_url,
87	AttributeNames=["FifoQueue"],
88	)
89	return get_queue_attributes_response.get("Attributes", {})	1✔
90
91	def event_source(self) -> str:	1✔
92	return "aws:sqs"	1✔
93
94	def poll_events(self) -> None:	1✔
95	# SQS pipe source: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
96	# "The 9 Ways an SQS Message can be Deleted": https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/
97	# TODO: implement batch window expires based on MaximumBatchingWindowInSeconds
98	# TODO: implement invocation payload size quota
99	# TODO: consider long-polling vs. short-polling trade-off. AWS uses long-polling:
100	# https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html#pipes-sqs-scaling
101	response = self.source_client.receive_message(	1✔
102	QueueUrl=self.queue_url,
103	MaxNumberOfMessages=min(self.batch_size, DEFAULT_MAX_RECEIVE_COUNT),
104	MessageAttributeNames=["All"],
105	MessageSystemAttributeNames=[MessageSystemAttributeName.All],
106	# Override how many messages we can receive per call
107	sqs_override_max_message_count=self.batch_size,
108	)
109	if messages := response.get("Messages"):	1✔
110	LOG.debug("Polled %d events from %s", len(messages), self.source_arn)	1✔
111	try:	1✔
112	if self.is_fifo_queue:	1✔
113	# TODO: think about starvation behavior because once failing message could block other groups
114	fifo_groups = split_by_message_group_id(messages)	1✔
115	for fifo_group_messages in fifo_groups.values():	1✔
116	self.handle_messages(fifo_group_messages)	1✔
117	else:
118	self.handle_messages(messages)	1✔
119
120	# TODO: unify exception handling across pollers: should we catch and raise?
121	except Exception as e:	1✔
122	# TODO: improve error messages (produce same failure and design better error messages)
123	LOG.warning(	1✔
124	"Polling or batch processing failed: %s",
125	e,
126	exc_info=LOG.isEnabledFor(logging.DEBUG),
127	)
128
129	def handle_messages(self, messages):	1✔
130	polled_events = transform_into_events(messages)	1✔
131	# Filtering: matching vs. discarded (i.e., not matching filter criteria)
132	# TODO: implement format detection behavior (e.g., for JSON body):
133	# https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
134	# Check whether we need poller-specific filter-preprocessing here without modifying the actual event!
135	# convert to json for filtering (HACK for fixing parity with v1 and getting regression tests passing)
136	for event in polled_events:	1✔
137	try:	1✔
138	event["body"] = json.loads(event["body"])	1✔
139	except json.JSONDecodeError:	1✔
140	LOG.debug(	1✔
141	"Unable to convert event body '%s' to json... Event might be dropped.",
142	event["body"],
143	)
144	matching_events = self.filter_events(polled_events)	1✔
145	# convert them back (HACK for fixing parity with v1 and getting regression tests passing)
146	for event in matching_events:	1✔
147	event["body"] = (	1✔
148	json.dumps(event["body"]) if not isinstance(event["body"], str) else event["body"]
149	)
150
151	all_message_ids = {message["MessageId"] for message in messages}	1✔
152	matching_message_ids = {event["messageId"] for event in matching_events}	1✔
153	discarded_message_ids = all_message_ids.difference(matching_message_ids)	1✔
154	# Delete discarded events immediately:
155	# https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/#7-event-source-mappings-with-filters
156	self.delete_messages(messages, discarded_message_ids)	1✔
157
158	# Don't trigger upon empty events
159	if len(matching_events) == 0:	1✔
UNCOV 160	return	×
161	# Enrich events with metadata after filtering
162	enriched_events = self.add_source_metadata(matching_events)	1✔
163
164	# Invoke the processor (e.g., Pipe, ESM) and handle partial batch failures
165	try:	1✔
166	self.processor.process_events_batch(enriched_events)	1✔
167	successful_message_ids = all_message_ids	1✔
168	except PartialBatchFailureError as e:	1✔
169	failed_message_ids = parse_batch_item_failures(	1✔
170	e.partial_failure_payload, matching_message_ids
171	)
172	successful_message_ids = matching_message_ids.difference(failed_message_ids)	1✔
173
174	# Only delete messages that are processed successfully as described here:
175	# https://docs.aws.amazon.com/en_gb/lambda/latest/dg/with-sqs.html
176	# When Lambda reads a batch, the messages stay in the queue but are hidden for the length of the queue's
177	# visibility timeout. If your function successfully processes the batch, Lambda deletes the messages
178	# from the queue. By default, if your function encounters an error while processing a batch,
179	# all messages in that batch become visible in the queue again. For this reason, your function code must
180	# be able to process the same message multiple times without unintended side effects.
181	# Troubleshooting: https://repost.aws/knowledge-center/lambda-sqs-report-batch-item-failures
182	# For FIFO queues, AWS also deletes successfully sent messages. Therefore, the AWS docs recommends:
183	# "If you're using this feature with a FIFO queue, your function should stop processing messages after the first
184	# failure and return all failed and unprocessed messages in batchItemFailures. This helps preserve the ordering
185	# of messages in your queue."
186	# Following this recommendation could result in the unsolved side effect that valid messages are continuously
187	# placed in the same batch as failing messages:
188	# * https://stackoverflow.com/questions/78694079/how-to-stop-fifo-sqs-messages-from-being-placed-in-a-batch-with-failing-messages
189	# * https://stackoverflow.com/questions/76912394/can-i-report-only-messages-from-failing-group-id-in-reportbatchitemfailures-resp
190
191	# TODO: Test blocking failure behavior for FIFO queues to guarantee strict ordering
192	# -> might require some checkpointing or retry control on the poller side?!
193	# The poller should only proceed processing FIFO queues after having retried failing messages:
194	# "If your pipe returns an error, the pipe attempts all retries on the affected messages before EventBridge
195	# receives additional messages from the same group."
196	# https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
197	self.delete_messages(messages, successful_message_ids)	1✔
198
199	def delete_messages(self, messages: list[dict], message_ids_to_delete: set):	1✔
200	"""Delete SQS `messages` from the source queue that match a MessageId within `message_ids_to_delete`"""
201	# TODO: unclear how (partial) failures for deleting are handled, retry or fail batch? Hard to test against AWS
202	if len(message_ids_to_delete) > 0:	1✔
203	entries = [	1✔
204	{"Id": str(count), "ReceiptHandle": message["ReceiptHandle"]}
205	for count, message in enumerate(messages)
206	if message["MessageId"] in message_ids_to_delete
207	]
208
209	self.source_client.delete_message_batch(	1✔
210	QueueUrl=self.queue_url,
211	Entries=entries,
212	# Override how many messages can be deleted at once
213	sqs_override_max_message_count=self.batch_size,
214	)
215
216
217	def split_by_message_group_id(messages) -> defaultdict[str, list[dict]]:	1✔
218	"""Splitting SQS messages by MessageGroupId to ensure strict ordering for FIFO queues"""
219	fifo_groups = defaultdict(list)	1✔
220	for message in messages:	1✔
221	message_group_id = message["Attributes"]["MessageGroupId"]	1✔
222	fifo_groups[message_group_id].append(message)	1✔
223	return fifo_groups	1✔
224
225
226	def transform_into_events(messages: list[dict]) -> list[dict]:	1✔
227	events = []	1✔
228	for message in messages:	1✔
229	# TODO: consolidate with SQS event source listener:
230	# localstack.services.lambda_.event_source_listeners.sqs_event_source_listener.SQSEventSourceListener._send_event_to_lambda
231	message_attrs = message_attributes_to_lower(message.get("MessageAttributes"))	1✔
232	event = {	1✔
233	# Original SQS message attributes
234	"messageId": message["MessageId"],
235	"receiptHandle": message["ReceiptHandle"],
236	# TODO: test with empty body
237	# TODO: implement heuristic based on content type: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
238	"body": message.get("Body", "MessageBody"),
239	"attributes": message.get("Attributes", {}),
240	"messageAttributes": message_attrs,
241	# TODO: test with empty body
242	"md5OfBody": message.get("MD5OfBody") or message.get("MD5OfMessageBody"),
243	}
244	# TODO: test Pipe with message attributes (only covered by Lambda ESM SQS test so far)
245	if md5_of_message_attributes := message.get("MD5OfMessageAttributes"):	1✔
246	event["md5OfMessageAttributes"] = md5_of_message_attributes	1✔
247	events.append(event)	1✔
248	return events	1✔
249
250
251	def get_queue_url(queue_arn: str) -> str:	1✔
252	# TODO: consolidate this method with localstack.services.sqs.models.SqsQueue.url
253	# * Do we need to support different endpoint strategies?
254	# * If so, how can we achieve this without having a request context
255	host_url = internal_service_url()	1✔
256	host = host_url.rstrip("/")	1✔
257	parsed_arn = parse_arn(queue_arn)	1✔
258	account_id = parsed_arn["account"]	1✔
259	name = parsed_arn["resource"]	1✔
260	return f"{host}/{account_id}/{name}"	1✔
261
262
263	def message_attributes_to_lower(message_attrs):	1✔
264	"""Convert message attribute details (first characters) to lower case (e.g., stringValue, dataType)."""
265	message_attrs = message_attrs or {}	1✔
266	for _, attr in message_attrs.items():	1✔
267	if not isinstance(attr, dict):	1✔
UNCOV 268	continue	×
269	for key, value in dict(attr).items():	1✔
270	attr[first_char_to_lower(key)] = attr.pop(key)	1✔
271	return message_attrs	1✔

localstack / localstack / f86a2f6e-a611-48ec-a078-fed1bbc27bef

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous