GCP Pub/Sub & Python - 如何从消息中获取 JSON 密钥？

Question

我必须首先道歉，因为我刚开始使用Pub/Sub，所以我可能对GCP功能缺乏了解。

我正在尝试调整谷歌自己的Python脚本，在Cloud Run中接收来自Pub/Sub的消息(https://cloud.google.com/run/docs/triggering/pubsub-push#run_pubsub_handler-python(：

@app.route("/", methods=["POST"])
def index():
envelope = request.get_json()
if not envelope:
msg = "no Pub/Sub message received"
print(f"error: {msg}")
return f"Bad Request: {msg}", 400
if not isinstance(envelope, dict) or "message" not in envelope:
msg = "invalid Pub/Sub message format"
print(f"error: {msg}")
return f"Bad Request: {msg}", 400
pubsub_message = envelope["message"]
name = "World"
if isinstance(pubsub_message, dict) and "data" in pubsub_message:
name = base64.b64decode(pubsub_message["data"]).decode("utf-8").strip()
print(f"Hello {name}!")
return ("", 204)

现在我使用的Pub/SubTopic有一个AVRO模式，比如：

{
"type": "record",
"name": "Avro",
"fields": [
{
"name": "ext_file_id",
"type": "string"
},
{
"name": "input_bucket",
"type": "string"
}
]
}

在Cloud Run(我的Python脚本(中，我需要做的是获取JSON主体键(ext_file_id和input_bucket(及其值。

考虑到我通过获得JSON

envelope = request.get_json()

它会简单地是这样的吗：

envlope.body.ext_file_id
envelope.body.input_bucket

还是别的什么？

我尝试了以下操作，但没有成功：

envelope["ext_file_id"]
envelope["input_bucket"]

Answer 1

当您从PubSub收到JSON时，您会收到一个包含此格式的带有message字段的JSON

{
"data": string,
"attributes": {
string: string,
...
},
"messageId": string,
"publishTime": string,
"orderingKey": string
}

您的消息在代码中的data字段中以base64编码

pubsub_message["data"]).decode("utf-8")

然后，您必须对该内容进行JSON解析(json.loads(data)(，然后获得所需的字段。