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

How to use message_deduplication_id method in localstack

Best Python code snippet using localstack_python

send_sqs_message_lambda.py

Source:send_sqs_message_lambda.py

1import json2import boto33from botocore.exceptions import ClientError4client = boto3.client('sqs')5def lambda_handler(event, context):6    sqsUrls = event['subscribers']7    message_body = event['message']8    delay_seconds = 09    if 'delay-seconds' in event:10        delay_seconds = event['delay-seconds']11    message_attributes = None12    if 'message-attributes' in event:13        message_attributes = event['message-attributes']14    message_deduplication_id = None15    if 'message-deduplication-id' in event:16        message_deduplication_id = event['message-deduplication-id']17    message_group_id = None18    if 'message-group-id' in event:19        message_group_id = event['message-group-id']20    21    failedAttempts = []22    23    for sqsUrl in sqsUrls:24        try:25            sendSQSMessage(26                sqsUrl, 27                message_body, 28                delay_seconds, 29                message_attributes,30                message_deduplication_id,31                message_group_id32                )33        except ClientError as e:34            failedAttempts.append({'sqsUrl': sqsUrl, 'error': e})35            36    if len(failedAttempts) > 0:37        raise Exception({'errorMessage': 'Could not send all SQS messages!', 'failedAttempts': failedAttempts})38        39def sendSQSMessage(sqsUrl, 40                message_body, 41                delay_seconds, 42                message_attributes,43                message_deduplication_id,44                message_group_id):45    message = constructMessage(sqsUrl, 46                            message_body, 47                            delay_seconds, 48                            message_attributes,49                            message_deduplication_id,50                            message_group_id)51    client.send_message(**message)52    53def constructMessage(sqsUrl, 54                message_body, 55                delay_seconds, 56                message_attributes,57                message_deduplication_id,58                message_group_id):59    message = {60        'QueueUrl': sqsUrl,61        'MessageBody': json.dumps(message_body)62    }63    if delay_seconds > 0:64        message.update({'DelaySeconds': delay_seconds})65    if message_attributes is not None:66        message.update({'MessageAttributes': message_attributes})67    if message_deduplication_id is not None:68        message.update({'MessageDeduplicationId': message_deduplication_id})69    if message_group_id is not None:70        message.update({'MessageGroupId': message_group_id})71    return message...

sqs_utils.py

Source:sqs_utils.py

1import boto32from datetime import datetime3def send_message(queue_url, message_body="none", message_attributes={}, message_group_id="message", message_deduplication_id=""):4    # Send message to SQS queue5    if not message_deduplication_id:6        message_deduplication_id = str(datetime.timestamp(datetime.now()))7    sqs = boto3.client('sqs')8    if not message_attributes:9        print("Change message_attributes")10        message_attributes = {11            'Title': {12                'DataType': 'String',13                'StringValue': 'NoTitle'14            },15            'Author': {16                'DataType': 'String',17                'StringValue': 'NoAuthor'18            }19        }20    21    return sqs.send_message(22            QueueUrl=queue_url,23            MessageDeduplicationId=message_deduplication_id,24            MessageAttributes=message_attributes,25            MessageGroupId=message_group_id,26            MessageBody=message_body27        )    28def receive_message(queue_url):29    # Receive message from SQS queue30    sqs = boto3.client('sqs')31    while 1:32        response = sqs.receive_message(33            QueueUrl=queue_url,34            AttributeNames=[35                'All'36            ],37            MaxNumberOfMessages=1,38            MessageAttributeNames=[39                'All'40            ],41            VisibilityTimeout=43200, # Max value42            WaitTimeSeconds=143        )44        if 'Messages' in response.keys():45            #print("Yes")46            break47        #else:48            #print("No")49    #print("[RESPONSE]="+str(response))50    message = response['Messages'][0]51    receipt_handle = message['ReceiptHandle']52    # Delete received message from queue53    sqs.delete_message(54        QueueUrl=queue_url,55        ReceiptHandle=receipt_handle56    )57    #print('Received and deleted message: %s' % message)58    return message...

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.