I'm trying to use the Google Cloud Tasks code sample provided in the GCP documentation here: https://cloud.google.com/tasks/docs/creating-http-target-tasks (see code below). However, it's unclear from the documentation which package I need for this to work. Any ideas?
However, I'm getting the following error.
ImportError: cannot import name 'tasks_v2' from 'google.cloud'
Code sample:
from google.cloud import tasks_v2
from google.protobuf import timestamp_pb2
# Create a client.
client = tasks_v2.CloudTasksClient()
# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# queue = 'my-appengine-queue'
# location = 'us-central1'
# payload = 'hello'
# Construct the fully qualified queue name.
parent = client.queue_path(project, location, queue)
# Construct the request body.
task = {
'app_engine_http_request': { # Specify the type of request.
'http_method': 'POST',
'relative_uri': '/example_task_handler'
}
}
if payload is not None:
# The API expects a payload of type bytes.
converted_payload = payload.encode()
# Add the payload to the request.
task['app_engine_http_request']['body'] = converted_payload
if in_seconds is not None:
# Convert "seconds from now" into an rfc3339 datetime string.
d = datetime.datetime.utcnow() + datetime.timedelta(seconds=in_seconds)
# Create Timestamp protobuf.
timestamp = timestamp_pb2.Timestamp()
timestamp.FromDatetime(d)
# Add the timestamp to the tasks.
task['schedule_time'] = timestamp
# Use the client to build and send the task.
response = client.create_task(parent, task)
print('Created task {}'.format(response.name))
return response
This worked for me:
pip3 install -U google-cloud-tasks
If you get any problems importing packages for other GCP services, this is a great resource: https://github.com/googleapis/google-cloud-python. The links point straight to the relevant package names on pypi.org.