Implementing a cloud function to publish to pubsub triggered by GCS finalize

Question

I've been trying to write and deploy a cloud function in Python. (gave up on node.js due to the messy documentation and relatively fast pace of changes)

It is meant to publish a message to a Pub/Sub topic, triggered when a file finishes being uploaded into google cloud bucket ("finalize").

The code I use to deploy the function is

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

I have been trying by using this script provided by 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)

To which I receive these errors in Google Cloud Console

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)

Following instruction from these two posts, I've copied the requirements.txt from the helloworld code sample, containing just

google-cloud-error-reporting==0.30.0

and updated other cloud functions like bigquery, storage, and logging. I then got these errors:

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`

and I also found [this thread](ImportError: cannot import name 'pubsub_v1' from 'google.cloud' (unknown location) but I don't really understand what's the solution, I've tried replacing pubsub_v1 with google-cloud-pubsub==0.38.0 which did not help. I get this error instead:

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

Also, this does not seem to be a sustainable fix if the code will break once Google updates pubsub to a new version?

So I'm very much a beginner and quite lost, but I hope this documentation could help you guys help me.

UPDATE:

It seems pubsub and pubsub_v1 are both valid to be used, not sure what's the difference, though.

@dustin I did a pip install -r requirements.txt which ended up matchiing what you provided. I've also noticed an error in deploying the function as 'hello-gcs-generic', which should be changed to 'callback'.

The python code now runs well locally, but deploying it to the cloud using the above code (the first code line in the OP) consistently returns this error

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

Answer 1

There is a simpler Python quickstart example that does what you need. ;-)

The example you are quoting is more advanced. It shows how to publish a message with error-handling. The while(True): sleep(60) line in the advanced example is there to keep the main thread alive, unless Ctrl+C or its equivalent is issued to stop the program from running. The reason for this sleep function to be there is so we can wait for the callback calls on the publish futures to finish and not exit the program immediately after the publish call. Again, this is perhaps a little too complex for what you are trying to learn to do with Cloud Pub/Sub with Cloud Functions. I would suggest to steer away from the advanced example and use the quickstart example.

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.')

Answer 2

You need to add google-cloud-pubsub to your requirements.txt file, not in your main.py file. It should look like this:

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