如何编写监控存储桶的云函数？答案

【问题标题】：How do I write a cloud function that monitors a storage bucket?如何编写监控存储桶的云函数？
【发布时间】：2020-01-15 06:47:31
【问题描述】：

我已经设置了一个 Google Cloud Storage 存储桶来向 Pub/Sub 主题发送通知：

gsutil notification create -t my-topic -f json gs://test-bucket

我已经创建了对该主题的订阅，以将消息推送到云功能端点：

gcloud pubsub subscriptions create my-sub --topic my-topic

并且云功能部署为：

gcloud functions deploy promo_received --region europe-west1 --runtime python37 --trigger-topic my-topic

该函数的目的（现在）是检查在测试桶中创建的文件是否与特定文件名匹配，并在匹配时向 Slack 发送消息。目前该函数如下所示：

def promo_received(data):
    date_str = datetime.today().strftime('%Y%m%d')
    filename = json.loads(data)["name"]
    bucket = json.loads(data)["bucket"]

    if filename == 'PROM_DTLS_{}.txt.gz'.format(date_str):
        msg = ":heavy_check_mark: *{}* has been uploaded to *{}*. Awaiting instructions.".format(filename, bucket)
        post_to_slack(url, msg)

当我通过删除一个名为 PROM_DTLS_20190913.txt.gz 的文件来测试它时，我可以看到函数触发了，但是它崩溃并出现 2 个错误：

TypeError: promo_received() takes 1 positional argument but 2 were given

TypeError: the JSON object must be str, bytes or bytearray, not LocalProxy

这是我第一次尝试这样做，我不知道从哪里开始进行故障排除。任何帮助将不胜感激！

【问题讨论】：

标签： python google-cloud-functions google-cloud-storage google-cloud-pubsub

【解决方案1】：

您需要为您的函数添加context 作为参数，这将解决第一个错误：

def promo_received(data, context):
    [...]

另外，您不需要json.loads 来检索文件或存储桶的名称：

data['name']
data['bucket']

这应该消除第二个错误。

查看Google Cloud Storage Triggers 文档页面中的示例

【讨论】：

感谢您的帮助 - 我已按照您的建议实施了更改，但现在我收到了 KeyError: 'name'。
@Cam 我是凭记忆说的，因为我现在无法运行测试，但你可以试试data.get('name') 吗？

【解决方案2】：

要编写 Python 云函数，请查看 this example。请注意 Cloud Storage serializes the object into a utf-8 JSON string，然后 Cloud Functions 对其进行 base64 编码。所以你需要先对payload进行base64解码，然后utf8解码，再进行json解析。

def promo_received(event, context):
  obj = json.loads(base64.b64decode(event['data']).decode('utf-8'))
  filename = obj[“name”]
  bucket = obj[“bucket”]

  # the rest of your code goes here

【讨论】：