GCP Pub/Sub + Python 如何从Cloud Run接收的消息中提取JSON键值

结论 #

你之前尝试的两种写法都不可行，正确的取值逻辑需要先处理Pub/Sub推送消息的固定封装结构。

错误写法原因说明 #

• Pub/Sub的推送请求顶级envelope结构仅包含系统级字段（如message、subscription等），你需要的业务字段并不在顶层，因此envelope["ext_file_id"]无法取到值
• request.get_json()返回的是Python字典对象，不存在body属性，因此envelope.body.ext_file_id的写法不符合Python字典的访问规则

正确处理逻辑 #

Pub/Sub推送的所有业务数据都经过base64编码后存放在envelope["message"]["data"]字段中，你需要按以下步骤处理：

1. 取出message对象中的data字段，执行base64解码
2. 将解码后的内容反序列化为Python字典，即可读取AVRO schema中定义的ext_file_id、input_bucket字段

场景1：Pub/Sub推送JSON编码的AVRO消息（默认通用场景） #

直接解码后转JSON即可，修改后的完整代码如下：

import json
import base64
from flask import Flask, request

app = Flask(__name__)

@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"]

    ext_file_id = ""
    input_bucket = ""
    if isinstance(pubsub_message, dict) and "data" in pubsub_message:
        # base64解码业务数据
        data_str = base64.b64decode(pubsub_message["data"]).decode("utf-8").strip()
        # 转为Python字典
        try:
            data_dict = json.loads(data_str)
            ext_file_id = data_dict.get("ext_file_id", "")
            input_bucket = data_dict.get("input_bucket", "")
        except json.JSONDecodeError:
            print("error: 无法解析消息数据为JSON格式")
            return "Bad Request: invalid message data", 400

    # 后续可直接使用两个字段处理业务逻辑
    print(f"ext_file_id: {ext_file_id}, input_bucket: {input_bucket}")

    return ("", 204)

场景2：Pub/Sub推送二进制格式的AVRO消息 #

如果你的订阅配置了推送二进制AVRO数据，需要额外使用avro库反序列化：

1. 先安装依赖
   pip install avro

2. 调整代码中的数据处理部分：

import avro.schema
from avro.io import DatumReader, BinaryDecoder
import io
import base64
from flask import Flask, request

app = Flask(__name__)

# 提前加载你定义的AVRO schema
avro_schema = avro.schema.parse("""
{
  "type": "record",
  "name": "Avro",
  "fields": [
    {"name": "ext_file_id", "type": "string"},
    {"name": "input_bucket", "type": "string"}
  ]
}
""")
avro_reader = DatumReader(avro_schema)

@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"]

    ext_file_id = ""
    input_bucket = ""
    if isinstance(pubsub_message, dict) and "data" in pubsub_message:
        # base64解码得到二进制AVRO数据
        data_bytes = base64.b64decode(pubsub_message["data"])
        # 反序列化二进制数据为字典
        bytes_reader = io.BytesIO(data_bytes)
        decoder = BinaryDecoder(bytes_reader)
        try:
            data_dict = avro_reader.read(decoder)
            ext_file_id = data_dict["ext_file_id"]
            input_bucket = data_dict["input_bucket"]
        except Exception as e:
            print(f"error: 反序列化AVRO数据失败: {str(e)}")
            return "Bad Request: invalid AVRO data", 400

    print(f"ext_file_id: {ext_file_id}, input_bucket: {input_bucket}")

    return ("", 204)

内容的提问来源于stack exchange，提问作者RobTheRobot16