json - 如何使用控制台在 aws SNS 中发送 json 消息正文
问题描述
我正在动手将 SNS 触发器添加到 lambda 函数,然后将消息发送到松弛通道。这个 lambda 在 python 中有一个蓝图,还有一个模板测试事件,如下所示
{
"Records": [
{
"EventVersion": "1.0",
"EventSubscriptionArn": "arn:aws:sns:EXAMPLE",
"EventSource": "aws:sns",
"Sns": {
"SignatureVersion": "1",
"Timestamp": "1970-01-01T00:00:00.000Z",
"Signature": "EXAMPLE",
"SigningCertUrl": "EXAMPLE",
"MessageId": "12345",
"Message": {
"AlarmName": "SlackAlarm",
"NewStateValue": "OK",
"NewStateReason": "Threshold Crossed: 1 datapoint (0.0) was not greater than or equal to the threshold (1.0)."
},
"MessageAttributes": {
"Test": {
"Type": "String",
"Value": "TestString"
},
"TestBinary": {
"Type": "Binary",
"Value": "TestBinary"
}
},
"Type": "Notification",
"UnsubscribeUrl": "EXAMPLE",
"TopicArn": "arn:aws:sns:EXAMPLE",
"Subject": "TestInvoke"
}
}
]
蓝图中的 lambda 处理程序中的代码如下
import boto3
import json
import logging
import os
from base64 import b64decode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
HOOK_URL = os.environ['kmsEncryptedHookUrl']
SLACK_CHANNEL = os.environ['slackChannel']
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info("Event: " + str(event))
message = event['Records'][0]['Sns']['Message']
logger.info("Message: " + str(message))
alarm_name = message['AlarmName']
new_state = message['NewStateValue']
reason = message['NewStateReason']
slack_message = {
'channel': SLACK_CHANNEL,
'text': "%s state is now %s: %s" % (alarm_name, new_state, reason)
}
req = Request(HOOK_URL, json.dumps(slack_message).encode('utf-8'))
try:
response = urlopen(req)
response.read()
logger.info("Message posted to %s", slack_message['channel'])
except HTTPError as e:
logger.error("Request failed: %d %s", e.code, e.reason)
except URLError as e:
logger.error("Server connection failed: %s", e.reason)
当我运行测试事件时,lambda 运行成功。
我想从控制台在 SNS 主题中发布一条消息,以查看 lambda 是否被正确触发。但是当我尝试将 JSON 对象作为消息体发布时,我收到了错误消息
[ERROR] TypeError: string indices must be integersTraceback (most recent call last): File "/var/task/lambda_function.py", line 21, in lambda_handler alarm_name = message['AlarmName']
我试着给普通的json
{
"AlarmName": "PublishedAlarm",
"NewStateValue": "OK",
"NewStateReason": "This alarm is published"
}
我尝试给出一个字符串化的 JSON
"{\"AlarmName\": \"PublishedAlarm\",\"NewStateValue\": \"OK\",\"NewStateReason\": \"This alarm is published\"}"
我尝试选择Custom payload for each delivery
消息结构,然后给出以下消息体
{
"default": "Sample fallback message",
"email": "Sample message for email endpoints",
"sqs": "Sample message for Amazon SQS endpoints",
"lambda": "{\"AlarmName\": \"PublishedAlarm\",\"NewStateValue\": \"OK\",\"NewStateReason\": \"This alarm is published\"}",
"http": "Sample message for HTTP endpoints",
"https": "Sample message for HTTPS endpoints",
"sms": "Sample message for SMS endpoints",
"firehose": "Sample message for Amazon Kinesis Data Firehose endpoints",
"APNS": "{\"aps\":{\"alert\": \"Sample message for iOS endpoints\"} }",
"APNS_SANDBOX": "{\"aps\":{\"alert\":\"Sample message for iOS development endpoints\"}}",
"APNS_VOIP": "{\"aps\":{\"alert\":\"Sample message for Apple VoIP endpoints\"}}",
"APNS_VOIP_SANDBOX": "{\"aps\":{\"alert\": \"Sample message for Apple VoIP development endpoints\"} }",
"MACOS": "{\"aps\":{\"alert\":\"Sample message for MacOS endpoints\"}}",
"MACOS_SANDBOX": "{\"aps\":{\"alert\": \"Sample message for MacOS development endpoints\"} }",
"GCM": "{ \"data\": { \"message\": \"Sample message for Android endpoints\" } }",
"ADM": "{ \"data\": { \"message\": \"Sample message for FireOS endpoints\" } }",
"BAIDU": "{\"title\":\"Sample message title\",\"description\":\"Sample message for Baidu endpoints\"}",
"MPNS": "<?xml version=\"1.0\" encoding=\"utf-8\"?><wp:Notification xmlns:wp=\"WPNotification\"><wp:Tile><wp:Count>ENTER COUNT</wp:Count><wp:Title>Sample message for Windows Phone 7+ endpoints</wp:Title></wp:Tile></wp:Notification>",
"WNS": "<badge version=\"1\" value=\"42\"/>"
}
没有任何效果。我还订阅了该主题的电子邮件地址,并且我收到的电子邮件没有任何问题。
如何模拟来自 SNS 的 lambda 事件模板中给出的测试事件?
解决方案
当您使用 SNS 发送纯 json 消息时,它将以以下格式传递给 lambda:
'Message': '{\n "AlarmName": "PublishedAlarm",\n "NewStateValue": "OK",\n "NewStateReason": "This alarm is published"\n}'
您可以使用ast '方法解析它literal_eval
:
import ast
#...
#...
def lambda_handler(event, context):
logger.info("Event: " + str(event))
message = event['Records'][0]['Sns']['Message']
logger.info("Message: " + str(message))
message = ast.literal_eval(event['Records'][0]['Sns']['Message'])
alarm_name = message['AlarmName']
new_state = message['NewStateValue']
reason = message['NewStateReason']
#...
#...
推荐阅读
- mobx - Mobx React Reactivity with Hooks 和 Observer
- ios - React-RCCText-framework 中的 Apple Mach-O 链接器错误
- angularjs - 在Angularjs中路由到不同的页面?
- python - 在python中将参数添加到url
- python - 如何使用预期条件 new_window_is_opened 切换选项卡并使用 Selenium 和 Python 验证页面标题
- powershell - Powershell Get-ChildItem 不为文件夹返回任何内容
- amazon-web-services - 没有看到使用 asyncio 和 aioboto3 进行 DynamoDB 写入的性能改进
- bash - 如何运行需要两种不同类型输入文件的命令(1000 次)
- android - 如何在我的应用中为将来的自动填充登录显示 Google 的用户和密码保护消息?
- python - 使用 Django 会话存储名称值以用于另一个视图