KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _prepare_messages_to_dlq method in localstack

Best Python code snippet using localstack_python

dead_letter_queue.py

Source:dead_letter_queue.py

...39def _send_to_dead_letter_queue(source_type, source_arn, dlq_arn, event, error):40    if not dlq_arn:41        return42    LOG.info("Sending failed execution %s to dead letter queue %s" % (source_arn, dlq_arn))43    messages = _prepare_messages_to_dlq(source_arn, event, error)44    if ":sqs:" in dlq_arn:45        queue_url = aws_stack.get_sqs_queue_url(dlq_arn)46        sqs_client = aws_stack.connect_to_service("sqs")47        error = None48        result_code = None49        try:50            result = sqs_client.send_message_batch(QueueUrl=queue_url, Entries=messages)51            result_code = result.get("ResponseMetadata", {}).get("HTTPStatusCode")52        except Exception as e:53            error = e54        if error or not result_code or result_code >= 400:55            msg = "Unable to send message to dead letter queue %s (code %s): %s" % (56                queue_url,57                result_code,58                error,59            )60            LOG.info(msg)61            raise Exception(msg)62    elif ":sns:" in dlq_arn:63        sns_client = aws_stack.connect_to_service("sns")64        for message in messages:65            sns_client.publish(66                TopicArn=dlq_arn,67                Message=message["MessageBody"],68                MessageAttributes=message["MessageAttributes"],69            )70    else:71        LOG.warning("Unsupported dead letter queue type: %s" % dlq_arn)72    return dlq_arn73def _prepare_messages_to_dlq(source_arn, event, error):74    messages = []75    custom_attrs = {76        "RequestID": {"DataType": "String", "StringValue": str(uuid.uuid4())},77        "ErrorCode": {"DataType": "String", "StringValue": "200"},78        "ErrorMessage": {"DataType": "String", "StringValue": str(error)},79    }80    if ":sqs:" in source_arn:81        custom_attrs["ErrorMessage"]["StringValue"] = str(error.result)82        for record in event.get("Records", []):83            msg_attrs = message_attributes_to_upper(record.get("messageAttributes"))84            message_attrs = {**msg_attrs, **custom_attrs}85            messages.append(86                {87                    "Id": record.get("messageId"),...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.