时间戳

首页 > 解决方案 > 如何为 Google Cloud Pubsub“创建”/“分配”日志处理程序？

python - 如何为 Google Cloud Pubsub“创建”/“分配”日志处理程序？

问题描述

从上一个线程的开发发现，提出问题时的假设是题外话（子进程实际上并没有引起问题），所以我正在做一个更有针对性的帖子。

我的错误信息：

找不到记录器“google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager”的处理程序

我的意图：

将 Google PubSub 消息属性作为 Python 变量传递，以便在以后的代码中重复使用。

我的代码：

import time
import logging

from google.cloud import pubsub_v1

project_id = "redacted"
subscription_name = "redacted"

def receive_messages_with_custom_attributes(project_id, subscription_name):
    """Receives messages from a pull subscription."""
    # [START pubsub_subscriber_sync_pull_custom_attributes]

    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(
        project_id, subscription_name)

    def callback(message):
        print('Received message: {}'.format(message.data))
        if message.attributes:
            #print('Attributes:')
            for key in message.attributes:
                value = message.attributes.get(key);
                #commented out to not print to terminal
                #which should not be necessary
                #print('{}: {}'.format(key, value))
        message.ack()

        print("this is before variables")
        dirpath = "~/subfolder1/"
        print(dirpath)
        namepath = message.data["name"]
        print(namepath)
        fullpath = dirpath + namepath
        print(fullpath)
        print("this is after variables")


    subscriber.subscribe(subscription_path, callback=callback)
    # The subscriber is non-blocking, so we must keep the main thread from
    # exiting to allow it to process messages in the background.
    print('Listening for messages on {}'.format(subscription_path))
    while True:
        time.sleep(60)
    # [END pubsub_subscriber_sync_pull_custom_attributes]

receive_messages_with_custom_attributes(project_id, subscription_name)

我运行上述代码的完整控制台输出：

Listening for messages on projects/[redacted]
Received message: {
  "kind": "storage#object",
  "id": "[redacted]/0.testing/1548033442364022",
  "selfLink": "https://www.googleapis.com/storage/v1/b/[redacted]/o/BSD%2F0.testing",
  "name": "BSD/0.testing",
  "bucket": "[redacted]",
  "generation": "1548033442364022",
  "metageneration": "1",
  "contentType": "application/octet-stream",
  "timeCreated": "2019-01-21T01:17:22.363Z",
  "updated": "2019-01-21T01:17:22.363Z",
  "storageClass": "MULTI_REGIONAL",
  "timeStorageClassUpdated": "2019-01-21T01:17:22.363Z",
  "size": "0",
  "md5Hash": "1B2M2Y8AsgTpgAmY7PhCfg==",
  "mediaLink": "https://www.googleapis.com/download/storage/v1/b/[redacted]/o/BSD%2F0.testing?generation=1548033442364022&alt=media",
  "crc32c": "AAAAAA==",
  "etag": "CPb0uvvZ/d8CEAE="
}

this is before variables
/home/[redacted]
No handlers could be found for logger "google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager"

如您所见，第一个字符串和字符串定义为变量已打印，但代码在尝试从刚刚生成的字典中定义变量时中断，并且没有print()执行进一步的 s。

可能相关的线程，该用户正在使用 cron 作业发布，并从 crontab envpaths 中找到了一个修复程序，但我的情况是接收而不使用任何 cron 作业，但可能暗示 python 后面/内部的另一层？

谁能帮我添加一个处理程序以使此代码按预期运行？

标签： pythonlogginggoogle-cloud-platformhandlergoogle-cloud-pubsub

解决方案

首先，如果我对您在输出中显示的内容理解正确，那么每当您对 Cloud Storage 对象进行更改时，您都会使用 Pub/Sub 通知发送消息。此信息可能会有所帮助。

现在，因为 message.data 是一个BYTES 对象message.data["name"]，所以不会工作。因此，不能作为字典索引。

要将其视为 dict，您首先必须将其解码为 base64 ( import base64)。之后，剩下的是一个看起来像 JSON 格式的字符串。然后，您使用json.load() （不要忘记import json）将此字符串转换为字典。现在您可以索引消息。

代码将是：

print("This is before variables")
dirpath = "/subfolder1/"
print(dirpath)

#Transform the bytes object into a string by decoding it
namepath = base64.b64decode(message.data).decode('utf-8')

#Transform the json formated string into a dict
namepath = json.loads(namepath)

print(namepath["name"])
fullpath = dirpath + namepath["name"]
print(fullpath)
print("this is after variables")

现在，如果您的意图是仅读取属性，则它们在顶部正确定义，例如：

    if message.attributes:
        print('Attributes:')
        for key in message.attributes:
            value = message.attributes.get(key)
            print('{}: {}'.format(key, value))

所以，你可以使用：

    print("this is before variables")
    dirpath = "~/subfolder1/"
    print(dirpath)
    namepath = message.attributes["objectId"]
    print(namepath)
    fullpath = dirpath + namepath
    print(fullpath)
    print("this is after variables")

请记住，对于这种特殊情况，"objectId"是文件的名称，因为它是来自 Cloud Storage 的 Pub/Sub 的通知使用的属性。如果您假装发送自定义消息，请更改"objectId"为您想要的属性名称。

推荐阅读