Day9 - SQS 串接 Notify 實作！
首先先建立一個consumer/資料夾，新增__init__.py以及notify_handler.py
在notify_handler.py輸入以下程式碼
import json
import requests
def line_notify_handler(event, context):
print(event)
body = json.loads(event['Records'][0]['body'])
print(body)
headers = {
'Authorization': f"Bearer {body['token']}",
'Content-Type': 'application/x-www-form-urlencoded'
}
r = requests.post('https://notify-api.line.me/api/notify',
headers=headers,
data={'message': body['message']})
print(r)
# AWS record sample
# {'Records': [{'messageId': 'fddc42ba-a122-4581-965e-d0144ac8a5ad', 'receiptHandle': 'AQEBjO32gY5pXOfOrmDR0hD4k1av9KyjbHFpc+rIBPV2Brif7Lo+jqnGevSjfFwlICyGf+BhWwKaxFw8XdB3QTzRbw0vnLURjnQeDSBrJHa/S57SRs9TOLRBq38maycAVg69iZbetg9VhLMBCcLtOtPHTzKkmo+/Sosm51WA5CzXK7A0rteikx6nxS1CUIpq6MAujodupP0Hgr5RjK5nH/nmxA4Db0leWEmLokalZbtlx4W14tp7PZxPOrQOLDaGrH//p4h32tY8IN3MkCqi+gyNT7kCU4KwCGOIrybb07ZWyKBTKw+KOMNr/Ykj4z2N1qxIvTM55UY9d8V29YsH32OjrZTei5P7Nke/51E2tWkmkqoFAlqzxDjQPvpP+Pvvr8aazeeZ6opkr59UefAiiyM71Q==', 'body': 'hi', 'attributes': {'ApproximateReceiveCount': '9', 'SentTimestamp': '1566621263072', 'SenderId': '901588721449', 'ApproximateFirstReceiveTimestamp': '1566621263072'}, 'messageAttributes': {}, 'md5OfBody': '49f68a5c8493ec2c0bf489821c21fc3b', 'eventSource': 'aws:sqs', 'eventSourceARN': 'arn:aws:sqs:us-east-1:901588721449:LINE_notify_consumer', 'awsRegion': 'us-east-1'}]}
接著在requirements.txt加入boto3，他是一個使用 python 介接 AWS 的套件
boto3==1.9.189
加入SQS_URL以及SQS_ARN到.env裡面
SQS_URL=sqs url
SQS_ARN=your sqs arn
add controller/notify_sqs_controller.py
from flask_restful import Resource, reqparse
import json
from lib.db import Database
import psycopg2.extras
import os
import boto3
def send_message(url, attr, body, delay=0):
cli.send_message(
QueueUrl=url,
DelaySeconds=0,
MessageAttributes=attr,
MessageBody=body,
)
class SendNotifyBySQSController(Resource):
cli = boto3.client("sqs", region_name=os.environ("region"))
def post(self):
parser = reqparse.RequestParser()
parser.add_argument(
'message', required=True, help='message can not be blank!')
args = parser.parse_args()
msg = args['message']
with Database() as db, db.connect() as conn:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
f"SELECT token FROM notify")
fetch = cur.fetchall()
for f in fetch:
body = {
'token': f"Bearer {f['token']}",
'message': f"Hello everyone, {msg}"
}
cli.send_message(
QueueUrl=os.environ("SQS_URL"),
DelaySeconds=0,
MessageAttributes={},
MessageBody=json.dumps(body),
)
return {'result': 'ok'}, 200
程式寫完了就是要加一條路由/notify/sqs
from controller.notify_sqs_controller import SendNotifyBySQSController
api.add_resource(SendNotifyBySQSController, '/notify/sqs')
接著透過wsgi在本地起一個 server
sls wsgi serve
再搭配 postman 來做測試，測試內容如下
{
"message": "test Content"
}
接著透過sls deploy部署上會遇到一個問題，會有 Access Denied，所以要在serverless.yml加入 IAM role 的設定
add iam in provider
iamRoleStatements:
- Effect: Allow
Action:
- sqs:SendMessage
Resource:
- ${env:SQS_ARN}
測試
結論
使用 SQS 這類服務都會需要透過boto3來幫忙串接，最需要注意的就是 IAM role，因為在本地端的 key 通常權限都是最大的，但上到 AWS 上就會有權限的問題，所以要記得加入 IAM 哦！
Code is here
專案也會持續更新，更多詳情可以 follow 我的專案 aws-python-line-api。
