Qiita Teams that are logged in
You are not logged in to any team

Log in to Qiita Team
Community
OrganizationEventAdvent CalendarQiitadon (β)
Service
Qiita JobsQiita ZineQiita Blog
1
Help us understand the problem. What are the problem?

More than 3 years have passed since last update.

Python から Amazon SNS のアプリケーションプラットフォームのエンドポイントへリクエストを発行してみる

0.はじめに

Python から Amazon SNS の アプリケーションプラットフォームのエンドポイントへ発行してみました。

前提条件は、以下。

  • 実行する IAM ユーザーが作成されていること。
  • Amazon SNS のプラットフォームアプリケーションが作成されていること。
  • 作成したプラットフォームアプリケーションにデバイスのエンドポイントが登録されていること。

※下の2つの前提条件については、別の記事で記載したいと思います。
 っていうか、「そっちを先に投稿しろよ。」って感じではあるんですが…。

1.AWS IAM ポリシーの作成

  1. AWS IAM のコンソールへアクセス。
  2. ダッシュボードの「ポリシーの作成」ボタンを押下。
    • FireShot Capture 189 - IAM Management Console_ - https___console.aws.amazon.com_iam_home.png

  3. 「ポリシーの作成」画面から「Visual Editor」タブを選択。
    • スクリーンショット 2017-12-20 15.08.05.png

  4. 「Visual Editor」で、以下の項目を入力し、「Resources」の「Add ARN」をクリック。
    • サービス : 「SNS」
    • Actions : 「Publish」

    • スクリーンショット 2017-12-20 15.09.15.png

  5. ダイアログが表示されるので、「Spacecify ARN for Topic」に Amazon SNS のプラットフォームアプリケーションのアプリケーションを入力し、「ADD」ボタンを押下。
    • スクリーンショット 2017-12-20 15.09.50.png

  6. 上記の手順で「Resources」の入力を終えたら、「Review policy」ボタンを押下。
    • スクリーンショット 2017-12-20 15.10.20.png

  7. 「ポリシーの作成」画面が表示されるので、以下の項目を入力し、「ポリシーの作成」ボタンを押下。
    • 名前 : ※任意

    • スクリーンショット 2017-12-20 15.10.42.png

2.AWS IAM ポリシーのアタッチ

  1. AWS IAM のコンソールへアクセス。
  2. 作成した IAM ポリシーの「ポリシー名」をクリック。
    • スクリーンショット 2017-12-20 15.10.58.png

  3. 「概要」画面の「アタッチされたエンティティ」タブを選択し、「アタッチ」ボタンを押下。
    • スクリーンショット 2017-12-20 15.11.14.png

  4. 「ポリシーのアタッチ」画面からアタッチするリソースを選択し、「ポリシーのアタッチ」ボタンを押下。
    • スクリーンショット 2017-12-20 15.11.36.png

  5. IAM ポリシーの設定を確認。
    • スクリーンショット 2017-12-20 15.11.55_.png

3.プログラム

  1. 以下の Python スクリプトを実行すれば、発行されます。
    • 34 行目 : accesskey = "[ACCESS-KEY]"
    • 35 行目 : secretkey = "[SECRET-KEY]"
    • 41 行目 : endpoint_arn = "[Endpoint-ARN]"

SampleSNS.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

# ---1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----
# ==============================================================================
#
# SampleSNS.py
#
#   * ex) $ ./SampleSNS.py
#
# ==============================================================================

import sys
import io
import logging
import json
import time

import datetime
from datetime import datetime as dt

import boto3
from boto3.session import Session

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
logging.basicConfig(level=logging.INFO, filename=(__file__ + ".log"), format="%(asctime)s %(levelname)s %(filename)s %(lineno)d %(funcName)s | %(message)s")

# ------------------------------------------------------------------------------
# Set
# ------------------------------------------------------------------------------
tmp_today = datetime.datetime.today()

# AWS
accesskey = "[ACCESS-KEY]"
secretkey = "[SECRET-KEY]"

region = "ap-northeast-1"
session = Session(aws_access_key_id=accesskey, aws_secret_access_key=secretkey, region_name=region)
client = session.client('sns')

endpoint_arn = "[Endpoint-ARN]"

# ------------------------------------------------------------------------------
# 「SNS — Boto 3 Docs 1.4.8 documentation」
# <http://boto3.readthedocs.io/en/latest/reference/services/sns.html#SNS.PlatformEndpoint.publish>
# ------------------------------------------------------------------------------
def PublishSnsEndpoint():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        title = 'Test - Amazon SNS - (' + tmp_today.strftime('%Y/%m/%d %H:%M:%S.%f') + ')'
        platform = endpoint_arn.split("/")[1]
        if platform == 'APNS' or platform == 'APNS_SANDBOX':
            # iOS
            dict = {'aps':{'alert':title,'sound':'default', 'badge':1}}
        elif platform == 'GCM':
            # Android
            dict = {'data':{'message':title}}
        else:
            logging.warn("Endpoint ARN Error. : [%s]", endpoint_arn)
            return

        #
        message = {platform:json.dumps(dict)}
        messageJSON = json.dumps(message)
        logging.info("messageJSON : [%s]", messageJSON)
        response = client.publish(
            TargetArn = endpoint_arn,
            Subject = 'サブジェクト',
            Message = messageJSON,
            MessageStructure = 'json')
        logging.info("response : [%s]", response)
    except Exception as e:
        logging.error("Type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ------------------------------------------------------------------------------
# Main
# ------------------------------------------------------------------------------
if __name__ == '__main__':
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)

    # Set
    logging.info("REGION : [%s]", region)

    # Args
    logging.info("Argc : [%d]", len(sys.argv))
    for i in range(len(sys.argv)):
        logging.info("Argv[%d] : [%s]", i, sys.argv[i])

    #
    PublishSnsEndpoint()

    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ---1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----

99.ハマりポイント

  • ハマるというほどではありませんが、Resources でやろうとしたらうまくいかなかったので、Clients で実行してます。

  • っていうか、今回前提条件として記載したこの辺がエラい苦労したんですが…。
    • Amazon SNS のプラットフォームアプリケーションが作成されていること。
    • 作成したプラットフォームアプリケーションにデバイスのエンドポイントが登録されていること。

  • それはまた別の記事で。

XX.まとめ

上でも記載してますが、
さっさと投稿して、こちらにも追記しないとな…。

Android 版が先かな。
iOS 何か大変…。

Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
1
Help us understand the problem. What are the problem?