Migration guide

tasks/app.yaml

@@ -0,0 +1,23 @@
+# Copyright 2019 Google LLC All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+runtime: python37
+service: worker
+
+# [START taskqueues_use_https]
+handlers:
+- url: /.*
+  script: auto
+  secure: always
+# [END taskqueues_use_https]

tasks/main.py

@@ -0,0 +1,64 @@
+# Copyright 2019 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# [START taskqueues_request_handler]
+from flask import Flask, request
+from google.cloud import datastore
+
+
+app = Flask(__name__)
+
+
+@app.route('/update_counter', methods=['POST'])
+def example_task_handler():
+    # [START taskqueues_secure_handler]
+    if request.headers.get('X-Appengine-Taskname') is None:
+        # You may use the presence of the X-Appengine-Taskname header to validate
+		# the request comes from Cloud Tasks.
+        print('Invalid Task: No X-Appengine-Taskname request header found')
+        return 'Bad Request - Invalid Task', 400
+    # [END taskqueues_secure_handler]
+
+    amount = int(request.get_data())
+    uodate_counter(amount)
+    return
+
+
+def update_counter(amount):
+    # Instantiates a client
+    client = datastore.Client()
+
+    key = client.key('Counter', 'count')
+    counter = client.get(key)
+
+    # Create entity if it doesn't exist
+    if counter is None:
+        counter = datastore.Entity(key)
+        previous = 0
+    else:
+        previous = counter['count']
+
+    counter.update({'count': amount + previous})
+    # Send update request
+    client.put(counter)
+
+    print('Counter: {}'.format(amount + previous))
+    return
+
+
+if __name__ == '__main__':
+    # This is used when running locally. Gunicorn is used to run the
+    # application on Google App Engine. See entrypoint in app.yaml.
+    app.run(host='127.0.0.1', port=8080, debug=True)
+# [END taskqueues_request_handler]

tasks/migration.py

@@ -0,0 +1,196 @@
+
+from google.cloud import tasks
+from google.protobuf import timestamp_pb2
+from google.protobuf import duration_pb2
+import json
+
+# Create a client.
+client = tasks.CloudTasksClient()
+
+project = 'my-project-id'
+location = 'us-central1'
+
+# [START taskqueues_using_yaml]
+parent = client.location_path(project, location)
+queue_blue = {
+    'name': client.queue_path(project, location, 'queue-blue'),
+    'rate_limits': {
+        'max_dispatches_per_second': 5
+    },
+    'app_engine_routing_override': {
+        'version': 'v2',
+        'service': 'task-module'
+    }
+}
+
+queue_red = {
+    'name': client.queue_path(project, location, 'queue-red'),
+    'rate_limits': {
+        'max_dispatches_per_second': 1
+    }
+}
+
+## Create Queue
+queues = [queue_blue, queue_red]
+for queue in queues:
+    response = client.create_queue(parent, queue)
+    print(response)
+# [END taskqueues_using_yaml]
+
+
+# [START taskqueues_processing_rate]
+# Get queue object
+queue_path = client.queue_path(project, location, 'queue-blue')
+queue = client.get_queue(queue_path)
+
+# Update queue object
+queue.rate_limits.max_dispatches_per_second = 20
+queue.rate_limits.max_concurrent_dispatches = 10
+
+# Send update request
+response = client.update_queue(queue)
+print(response)
+# [END taskqueues_processing_rate]
+
+
+# [START taskqueues_new_task]
+amount = '10'.encode()
+queue = 'default'
+
+# Construct the fully qualified queue name.
+parent = client.queue_path(project, location, queue)
+
+# Construct the request body.
+task = {
+    'app_engine_http_request': {
+        'http_method': 'POST',
+        'relative_uri': '/update_counter',
+        'app_engine_routing': {
+            'service': 'worker'
+        },
+        'body': amount
+    }
+}
+
+# Use the client to build and send the task.
+response = client.create_task(parent, task)
+
+eta = response.schedule_time.ToDatetime().strftime("%m/%d/%Y, %H:%M:%S")
+print('Task {} enqueued, ETA {}.'.format(response.name, eta))
+# [END taskqueues_new_task]
+
+# [START taskqueues_passing_data]
+task1 = {
+    'app_engine_http_request': {
+        'http_method': 'POST',
+        'relative_uri': '/update_counter?key=blue',
+        'app_engine_routing': {
+            'service': 'worker'
+        }
+    }
+}
+
+task2 = {
+    'app_engine_http_request': {
+        'http_method': 'POST',
+        'relative_uri': '/update_counter',
+        'app_engine_routing': {
+            'service': 'worker'
+        },
+        'headers': {
+            'Content-Type': 'application/json'
+        },
+        'body': json.dumps({'key': 'blue'}).encode()
+    }
+}
+
+# Use the client to build and send the task.
+tasks = [task1, task2]
+for task in tasks:
+    response = client.create_task(parent, task)
+    print(response)
+# [END taskqueues_passing_data]
+
+# [START taskqueues_naming_tasks]
+task = {
+    'name': client.task_path(project, location, queue, 'first-try'),
+    'app_engine_http_request': {
+        'http_method': 'GET',
+        'relative_uri': '/url/path'
+    }
+}
+response = client.create_task(parent, task)
+print(response)
+# [END taskqueues_naming_tasks]
+
+# [START taskqueues_deleting_tasks]
+task_path = client.task_path(project, location, 'queue1', 'foo')
+response = client.delete_task(task_path)
+# [END taskqueues_deleting_tasks]
+
+# [START taskqueues_purging_tasks]
+queue_path = client.queue_path(project, location, 'queue1')
+response = client.purge_queue(queue_path)
+# [END taskqueues_purging_tasks]
+
+# [START taskqueues_pause_queue]
+queue_path = client.queue_path(project, location, 'queue1')
+response = client.pause_queue(queue_path)
+# [END taskqueues_pause_queues]
+
+# [START taskqueues_deleting_queues]
+queue_path = client.queue_path(project, location, 'queue1')
+response = client.delete_queue(queue_path)
+# [END taskqueues_deleting_queues]
+
+# [START taskqueues_retrying_tasks]
+from google.protobuf import duration_pb2
+
+fooqueue = {
+    'name': client.queue_path(project, location, 'fooqueue'),
+    'rate_limits': {
+        'max_dispatches_per_second': 1
+    },
+    'retry_config': {
+        'max_attempts': 7,
+        'max_retry_duration': 2*60*60*24
+    }
+}
+
+min = duration_pb2.Duration()
+min.seconds = 10
+
+max = duration_pb2.Duration()
+max.seconds = 200
+
+barqueue = {
+    'name': client.queue_path(project, location, 'barqueue'),
+    'rate_limits': {
+        'max_dispatches_per_second': 1
+    },
+    'retry_config': {
+        'min_backoff': min,
+        'max_backoff': max,
+        'nax_doublings': 0
+    }
+}
+
+max.seconds = 300
+bazqueue = {
+    'name': client.queue_path(project, location, 'bazqueue'),
+    'rate_limits': {
+        'max_dispatches_per_second': 1
+    },
+    'retry_config': {
+        'min_backoff': min,
+        'max_backoff': max,
+        'nax_doublings': 3
+    }
+}
+
+## Create Queue
+queues = [fooqueue, barqueue, bazqueue]
+for queue in queues:
+    response = client.create_queue(parent, queue)
+    print(response)
+# [END taskqueues_retrying_tasks]

tasks/requirements.txt

@@ -2,3 +2,4 @@ Flask==1.0.2
 gunicorn==19.9.0
 google-cloud-tasks==1.1.0
 googleapis-common-protos==1.6.0
+google-cloud-datastore==1.7.3