producer.py

producer.py

import os
import boto3
import json
import logging
from mypy_boto3_sqs.client import SQSClient

logging.basicConfig(level=logging.INFO)
queue_url = os.environ["MY_QUEUE_URL"]
client: SQSClient = boto3.client("sqs")

def send_message(no: int):
    result = client.send_message(
        QueueUrl=queue_url,
        MessageGroupId="my-group",
        MessageDeduplicationId=f"{no}",
        MessageBody=json.dumps({"no": no}),
    )
    logging.info("Sent message: no:%s\tMsgId:%s", no, result['MessageId'])

if __name__ == "__main__":
    logging.info("Start to put SQS messages.")

    for n in range(10):
        send_message(no=n+1)

    logging.info("End to put SQS messages.")