LambdaとSNSでごみ収集日をメール通知 #lambda

はじめに

平日休みだと、日課のごみ出しを忘れてしまうことがあるので、前日に通知しておきたいと思いました。
せっかくなのでLambda(Python)とSNSを使ってメール通知したいと思います。

今回構成するサービス

「毎日20時に翌日のごみ収集内容をEメール通知する」というサービスを作ってみようと思います。

大まかな流れは以下の流れで行います。
①SNSでTopicとサブスクリプションの作成
②Topicに対し、テストメッセージを送る
③IAMロールの作成
④Lamdaでテストコードの実行
⑤明日の曜日を判定するコードの作成
⑥SNSでメッセージの送信
⑦Lambda関数の定期的な実行

SNSでTopicとサブスクリプションの作成

SNSのサービス画面でTopicを作成します。

画面上の指示に従い、以下のようにTopicを作成します。
・タイプは「スタンダード」
・名前は「SendMail」

Topicが作成できたら次はサブスクリプションの作成をします。

サブスクリプションも画面上の指示に従い、以下のように作成します。
・トピックARNは自動入力されるが、あとで使用するため控えておきます。
・プロトコルは「Eメール」
・エンドポイントの部分にサブスクリプションの対象にしたいアドレスを入力します。

サブスクリプションを作成すると、対象にしたアドレスに対し、確認メールが飛びます。
この時点ではエンドポイントはID未付与でステータスが保留中のステータスになっています。

確認メールのリンクをクリックすると、確認が完了し、ステータスが確認済みに変わります。

これでSNSで配信ができる状態になりました。

Topicに対し、テストメッセージを送る

登録したアドレスに対してテストメッセージを送信したいと思います。

「メッセージの発行」をクリックする。

件名、メッセージ本文を入力し、「メッセージの発行」をクリックします。
サブスクリプションに登録したアドレスに対してメールが発行されていれば成功です。

IAMロールの作成

IAMサービスで、SNSのpublish権限を持つポリシーを作成し、Lambdaサービスのためのロールに紐づけます。
ロールの作成方法は、前回の記事を参照。

Lambdaでテストコードの実行

次はLambdaの関数を作成します。
今回はPythonでコードを記述します。

Lambdaのサービス画面で「関数の作成」をクリックします。

画面の指示に従い、以下のような設定で関数を作成します。
・オプションは「一から作成」
・関数名は「SendMailFunction」
・ランタイムは「Python 3.8」
・実行ロールに先ほど作成したロールを指定

「関数の作成」をクリックすると関数が作成されます。

コードソース欄でlambda_function.pyファイルを右クリックして「Open」を選択します。
そうすると、lambda_handlerという関数が既定で定義されています。
この関数に実装したい処理を記載します。

まずはテストコードを実行したいので、以下のように編集します。

import json

def lambda_handler(event, context):
    # TODO implement
    print('test')
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

編集したら、「Deploy」をクリックします。
「Changes deployed」が表示されれば成功です。

デプロイしたコードをテストするため、「テスト」をクリックます。
「新しいイベント」を選択し、「呼び出し」をクリックします。

上部に「実行結果：成功(ログ)」と表示されるので、詳細をクリックします。
ログ出力に「test」というメッセージが表示が表示されていれば成功です。

明日の曜日を判定するコードの作成

SNSでメールを送る処理する前に、明日の曜日を判定する処理を実装してみます。

import json
import datetime as dt

def lambda_handler(event, context):
    # TODO implement
    weekday = dt.datetime.now() + dt.timedelta(hours=33)
    msg = ''

    if weekday.weekday() == 0:
        msg = '明日は燃やせるごみ'
    elif weekday.weekday() == 1:
        msg = '明日はプラスチックごみ'
    elif weekday.weekday() == 3:
        msg = '明日は燃やせるごみ'
    elif weekday.weekday() == 4:
        msg = '明日は資源ごみ'

    print(msg)

    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

日付の判定にはdatetimeモジュールを使用しました。
ポイントは、LambdaはUTC基準なので、日本時刻に合わせるために9時間補正する必要があります。
そのため、現在時刻に24時間+9時間=33時間加算した時刻の曜日を判定しています。

SNSでメッセージの送信

以下のようなコードを作成しました。

import json
import datetime as dt
import boto3

def lambda_handler(event, context):
    # TODO implement
    weekday = dt.datetime.now() + dt.timedelta(hours=33)
    msg = ''
    topic_arn = '<TopicのARN>'
    subject = '明日のごみ収集内容'

    if weekday.weekday() == 0:
        msg = '明日は燃やせるごみ'
    elif weekday.weekday() == 1:
        msg = '明日はプラスチックごみ'
    elif weekday.weekday() == 3:
        msg = '明日は燃やせるごみ'
    elif weekday.weekday() == 4:
        msg = '明日は資源ごみ'

    client = boto3.client('sns')

    request = {
        'TopicArn' : topic_arn,
        'Message' : msg,
        'Subject' : subject
    }

    response = client.publish(**request)

    print(response)

    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

LambdaからSNSを使用するために、boto3モジュールを使用しています。
TopicのARN、メッセージ本文、Subjectを指定ます。
テスト実行で、登録したアドレスに対してメールが送信されれば成功です。

Lambda関数の定期的な実行

あとは作成した関数を毎日実行させてやればOKです。

作成した関数の詳細画面を呼び出し、「設定」->「トリガー」を選択します。
「トリガーを追加」をクリックし、トリガーとして「EventBridge(CloudWatch Events)」を選びます。
「新規ルールの作成」を選択し、
・ルール名を「daily_mail」
・ルールタイプを「スケジュール式」
・スケジュール式に「cron(0 11 * * ? *)」(毎日20時(UTCから9時間補正)に実行)
・トリガーの有効化にチェック

指定した時間にメールが発行されれば成功です。