LambdaでSQSにメッセージを登録

SQSにダミーメッセージを登録する方法

SQSをプロジェクトで活用しており、
ダミーデータを手動で登録するのが面倒なので、Lambdaを活用したデータ登録を行いました。

簡単なイメージ

コード

import json
import boto3
import logging
from datetime import datetime
from typing import List, Dict

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# SQSクライアントの初期化
sqs = boto3.client("sqs")

# SQSキューのURL（定数として定義）
QUEUE_URL: str = "https://sqs.ap-northeast-1.amazonaws.com/xxxxxxxxxxx/xxx"

# 1回の送信バッチ数（SQSの上限は10件）
BATCH_SIZE: int = 10


def lambda_handler(event: Dict, context: Dict) -> Dict[str, str]:
    """
    Lambda のエントリーポイント
    """
    num_messages: int = 1000  # 送信するメッセージの数
    send_messages_to_sqs(num_messages)
    return {"statusCode": 200, "body": json.dumps("Messages sent successfully!")}


def generate_dummy_data(index: int) -> Dict[str, str]:
    """
    ダミーデータを生成する関数
    """
    return {
        "url": "https://test.com/index",
        "title": f"[test]{index}",
        "contents": "テストです",
        "id": index,
        "created_at": datetime.utcnow().isoformat() + "Z",
    }


def send_messages_to_sqs(num_messages: int) -> None:
    """
    SQS にメッセージを送信する関数（バッチ処理）
    """
    messages: List[Dict[str, str]] = [
        {"Id": str(i), "MessageBody": json.dumps(generate_dummy_data(i))}
        for i in range(1, num_messages + 1)
    ]
    send_messages_in_batches(messages)


def send_messages_in_batches(messages: List[Dict[str, str]]) -> None:
    """
    SQS へバッチ送信する関数
    """
    for i in range(0, len(messages), BATCH_SIZE):
        batch: List[Dict[str, str]] = messages[i : i + BATCH_SIZE]  # 10件ずつ送信
        response = sqs.send_message_batch(QueueUrl=QUEUE_URL, Entries=batch)
        logger.info(f"Batch sent successfully: {response}")

lambda_handler

概要

Lambda 関数のエントリーポイントです。
Lambda が呼び出されると最初に実行されます。
この関数は、指定された数のメッセージを SQS（Simple Queue Service）に送信するための処理を呼び出します。

引数

• event: Lambda がトリガーされたときに送られてくるイベント情報（Dict 型）。
• context: Lambda 実行環境に関する情報を含むコンテキスト（Dict 型）。

戻り値

• HTTP ステータスコード 200 とメッセージの JSON 文字列を返します。

処理内容

• num_messages 変数で指定された数のメッセージを SQS に送信する処理を呼び出します。

generate_dummy_data

概要

ダミーデータを生成するための関数です。
テストやデモンストレーションのために、指定されたインデックスをもとにユニークなデータを作成します。

引数

• index: ダミーデータの識別子として使われる整数（int 型）。

戻り値

• ダミーデータを格納した辞書（Dict[str, str] 型）。URL、タイトル、内容、ID、作成日時などを含みます。

処理内容

• index を基にしたダミーデータを生成し、辞書として返します。生成されるデータは、テスト用の URL、タイトル、内容、ID、作成日時（ISO 8601形式）を含みます。

send_messages_to_sqs

概要

指定された数のメッセージを SQS に送信するための関数です。
この関数は、generate_dummy_data を使用してダミーデータを生成し、メッセージをまとめて SQS に送信します。

引数

• num_messages: 送信するメッセージの数（int 型）。

戻り値

• 戻り値は None です。この関数はメッセージを送信する副作用を持ち、明示的な戻り値はありません。

処理内容

• 引数として渡された num_messages 数に基づいて、メッセージを生成します。生成したメッセージは send_messages_in_batches 関数を使って、バッチ処理で SQS に送信されます。

send_messages_in_batches

概要

複数のメッセージをバッチ単位で SQS に送信する関数です。AWS SQS の制限により、1回の送信で最大 10 件のメッセージしか送信できないため、この関数ではメッセージを 10 件ずつのバッチに分けて送信します。

引数

• messages: 送信するメッセージのリスト（List[Dict[str, str]] 型）。各メッセージは辞書形式で格納されており、Id と MessageBody を含みます。

戻り値

• 戻り値は None です。この関数も副作用があり、メッセージを SQS に送信します。

処理内容

• メッセージを 10 件ずつに分割して、send_message_batch メソッドを用いて SQS にバッチ送信します。送信結果はログとして出力されます。

まとめ

Lambda を利用して、ダミーデータを指定された数だけ生成し、AWS SQS に送信する一連の処理を実行します。
ダミーデータは、簡単にテストやデモ用のデータを作成するために利用でき、SQS への送信処理はバッチ処理を使用して効率的に行われます。

SAMテンプレート

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: 'sample SAM Template for create-sample-log '
Parameters:
  Environment:
    Type: String
  SampleFunctionRoleName:
    Type: String
Resources:
  CreateSampleLogFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName:
        Fn::Sub: create-sample-log
      CodeUri: functions/create-sample-log/
      Handler: app.lambda_handler
      Runtime: python3.11
      Timeout: 60
      MemorySize: 128
      Architectures:
      - x86_64
      Environment:
        Variables:
          Environment:
            Ref: Environment
      Role:
        Fn::GetAtt:
        - SampleFunctionRole
    Metadata:
      SamResourceId: CreateSampleLogFunction
  SampleFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName:
        Ref: SampleFunctionRoleName
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
          Action:
          - sts:AssumeRole
      ManagedPolicyArns:
      - Ref: LambdaExecutionPolicy
    Metadata:
      SamResourceId: SampleFunctionRole
Outputs:
  CreateSampleLogFunction:
    Description: CreateSampleLog Lambda Function ARN
    Value:
      Fn::GetAtt:
      - CreateSampleLogFunction
      - Arn