实现云功能以发布到GCS finalize触发的pubsub

Question

我一直试图用Python编写和部署一个云函数。(由于文档混乱和更改速度相对较快，所以放弃了node.js )

它的目的是将消息发布到Pub/Sub主题，当文件上传到GoogleCloud桶(“finishes”)时触发消息。

用于部署该函数的代码是

gcloud functions deploy hello_gcs_generic --runtime python37 --trigger-resource bucketcfpubsub

我一直在尝试使用此脚本由Google提供

import time

from google.cloud import pubsub_v1

project_id = "bucketcfpubsub"
topic_name = "projects/bucketcfpubsub/topics/pubsub"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)

def callback(message_future):
# When timeout is unspecified, the exception method waits indefinitely.
if message_future.exception(timeout=30):
    print('Publishing message on {} threw an Exception {}.'.format(
        topic_name, message_future.exception()))
else:
    print(message_future.result())

for n in range(1, 10):
    data = u'Message number {}'.format(n)
# Data must be a bytestring
    data = data.encode('utf-8')
# When you publish a message, the client returns a Future.
    message_future = publisher.publish(topic_path, data=data)
    message_future.add_done_callback(callback)

print('Published message IDs:')

# We must keep the main thread from exiting to allow it to process
# messages in the background.
while True:
    time.sleep(60)

在Google控制台中收到这些错误的

ERROR: (gcloud.functions.deploy) OperationError: code=3, message=Function load error: Code in file main.py can't be loaded.
Detailed stack trace: Traceback (most recent call last):
  File "/env/local/lib/python3.7/site-packages/google/cloud/functions_v1beta2/worker.py", line 256, in check_or_load_user_function
    _function_handler.load_user_function()
  File "/env/local/lib/python3.7/site-packages/google/cloud/functions_v1beta2/worker.py", line 166, in load_user_function
    spec.loader.exec_module(main)
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/user_code/main.py", line 3, in <module>
    from google.cloud import pubsub_v1
ImportError: cannot import name 'pubsub_v1' from 'google.cloud' (unknown location)

按照这些 二贴子的指示，我从helloworld代码示例中复制了requirements.txt，只包含以下内容

google-cloud-error-reporting==0.30.0

并更新其他云功能，如bigquery、存储和日志记录。然后我得到了以下错误：

ERROR: (gcloud.functions.deploy) OperationError: code=3, message=Function load error: Code in file main.py can't be loaded.
Detailed stack trace: Traceback (most recent call last):
  File "/env/local/lib/python3.7/site-packages/google/cloud/functions_v1beta2/worker.py", line 256, in check_or_load_user_function
    _function_handler.load_user_function()
  File "/env/local/lib/python3.7/site-packages/google/cloud/functions_v1beta2/worker.py", line 166, in load_user_function
    spec.loader.exec_module(main)
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/user_code/main.py", line 3, in <module>
from google.cloud import pubsub_v1`

我还找到了这个线程(v1‘从'google.cloud’(未知位置)，但我真的不明白解决方案是什么)，我尝试用google==0.38.0代替pubsub_v1，但这并没有帮助。相反，我得到了这个错误：

Deploying function (may take a while - up to 2 minutes)...failed.
ERROR: (gcloud.functions.deploy) OperationError: code=3, message=Function load error: Code in file main.py can't be loaded.
Detailed stack trace: Traceback (most recent call last):
  File "/env/local/lib/python3.7/site-packages/google/cloud/functions_v1beta2/worker.py", line 256, in check_or_load_user_function
    _function_handler.load_user_function()
  File "/env/local/lib/python3.7/site-packages/google/cloud/functions_v1beta2/worker.py", line 166, in load_user_function
    spec.loader.exec_module(main)
  File "<frozen importlib._bootstrap_external>", line 724, in exec_module
  File "<frozen importlib._bootstrap_external>", line 860, in get_code
  File "<frozen importlib._bootstrap_external>", line 791, in source_to_code
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/user_code/main.py", line 3

此外，这似乎不是一个可持续的修复，如果代码将打破一旦谷歌更新发布到一个新版本？

因此，我是一个初学者，相当迷茫，但我希望这些文档可以帮助你们帮助我。

更新：

看起来pubsub和pubsub_v1都是有效的，但不确定有什么不同。

@dustin，我做了一个pip安装-r requirements.txt，最终与您提供的内容相匹配。我还注意到在将函数部署为“hello-gcs-泛型”时出现了错误，应该将其更改为“回调”。

python代码现在在本地运行良好，但是使用上面的代码(OP中的第一行代码)将其部署到云中，始终返回此错误。

ERROR: (gcloud.functions.deploy) OperationError: code=3, messa
ge=Function load error: Error: function load attempt timed out
.

Answer 1

您需要将google-cloud-pubsub添加到requirements.txt文件中，而不是在main.py文件中。它应该是这样的：

google-cloud-error-reporting==0.30.0
google-cloud-pubsub==0.38.0

Answer 2

有一个更简单的Python快速启动示例可以满足您的需要。;-)

您引用的示例更高级。它展示了如何发布带有错误处理的消息。高级示例中的while(True): sleep(60)行是为了保持主线程活动，除非发出Ctrl+C或其等效的命令来阻止程序运行。这个sleep函数存在的原因是，我们可以等待发布期货上的回调调用完成，而不是在发布调用之后立即退出程序。同样，对于您尝试学习使用Cloud函数的Cloud /Sub来说，这可能有点太复杂了。我建议避开高级示例，使用快速启动示例。

from google.cloud import pubsub_v1

# TODO project_id = "Your Google Cloud Project ID"
# TODO topic_name = "Your Pub/Sub topic name"

publisher = pubsub_v1.PublisherClient()
# The `topic_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/topics/{topic_name}`
topic_path = publisher.topic_path(project_id, topic_name)

for n in range(1, 10):
    data = u'Message number {}'.format(n)
    # Data must be a bytestring
    data = data.encode('utf-8')
    # When you publish a message, the client returns a future.
    future = publisher.publish(topic_path, data=data)
    print('Published {} of message ID {}.'.format(data, future.result()))

print('Published messages.')