add python snippets and tests for creating, listing, and deleting queues

tasks/create_queue.py

@@ -0,0 +1,35 @@
+# Copyright 2020 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 cloud_tasks_create_queue]
+def create_queue(project, queue_name, location):
+    """Create a task queue."""
+
+    from google.cloud import tasks_v2
+
+    # Create a client.
+    client = tasks_v2.CloudTasksClient()
+
+    # Construct the fully qualified location path.
+    parent = client.location_path(project, location)
+
+    # Construct the create queue request.
+    queue = {'name': client.queue_path(project, location, queue_name)}
+
+    # Use the client to create the queue.
+    response = client.create_queue(parent, queue)
+
+    print('Created queue {}'.format(response.name))
+    return response
+# [END cloud_tasks_create_queue]

tasks/create_queue_test.py

@@ -0,0 +1,40 @@
+# Copyright 2020 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.
+
+import os
+import uuid
+
+from google.cloud import tasks_v2
+import pytest
+
+import create_queue
+
+TEST_PROJECT_ID = os.environ['GOOGLE_CLOUD_PROJECT']
+TEST_LOCATION = os.getenv('TEST_QUEUE_LOCATION', 'us-central1')
+TEST_QUEUE_NAME = f'my-queue-{uuid.uuid4().hex}'
+
+
+@pytest.fixture()
+def test_queue():
+    client = tasks_v2.CloudTasksClient()
+    q = create_queue.create_queue(TEST_PROJECT_ID, TEST_QUEUE_NAME, TEST_LOCATION)
+
+    yield q
+
+    client.delete_queue(q.name)
+
+
+def test_create_queue(capsys, test_queue):
+    out, _ = capsys.readouterr()
+    assert 'Created queue' in out

tasks/delete_queue.py

@@ -0,0 +1,30 @@
+# Copyright 2020 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 cloud_tasks_delete_queue]
+def delete_queue(project, queue_name, location):
+    """Delete a task queue."""
+
+    from google.cloud import tasks_v2
+
+    # Create a client.
+    client = tasks_v2.CloudTasksClient()
+
+    # Get the fully qualified path to queue.
+    queue = client.queue_path(project, location, queue_name)
+
+    # Use the client to delete the queue.
+    client.delete_queue(queue)
+    print('Deleted queue')
+# [END cloud_tasks_delete_queue]

tasks/delete_queue_test.py

@@ -0,0 +1,55 @@
+# Copyright 2020 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.
+
+import os
+import uuid
+
+from google.cloud import exceptions, tasks_v2
+import pytest
+
+import delete_queue
+
+
+TEST_PROJECT_ID = os.environ['GOOGLE_CLOUD_PROJECT']
+TEST_LOCATION = os.getenv('TEST_QUEUE_LOCATION', 'us-central1')
+TEST_QUEUE_NAME = f'my-queue-{uuid.uuid4().hex}'
+
+
+@pytest.fixture()
+def test_queue():
+    client = tasks_v2.CloudTasksClient()
+    parent = client.location_path(TEST_PROJECT_ID, TEST_LOCATION)
+    queue = {
+        # The fully qualified path to the queue
+        'name': client.queue_path(
+            TEST_PROJECT_ID, TEST_LOCATION, TEST_QUEUE_NAME),
+    }
+    q = client.create_queue(parent, queue)
+
+    yield q
+
+    try:
+        # Attempt to delete the queue in case the sample failed.
+        client.delete_queue(q.name)
+    except exceptions.NotFound:
+        # The queue was already successfully deleted.
+        print('Queue already deleted successfully')
+
+
+def test_delete_queue(capsys, test_queue):
+    delete_queue.delete_queue(
+        TEST_PROJECT_ID, TEST_QUEUE_NAME, TEST_LOCATION
+    )
+    out, _ = capsys.readouterr()
+    assert 'Deleted queue' in out

tasks/list_queues.py

@@ -0,0 +1,36 @@
+# Copyright 2020 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 cloud_tasks_list_queues]
+def list_queues(project, location):
+    """List all task queues."""
+
+    from google.cloud import tasks_v2
+
+    # Create a client.
+    client = tasks_v2.CloudTasksClient()
+
+    # Construct the fully qualified location path.
+    parent = client.location_path(project, location)
+
+    # Use the client to obtain the queues.
+    response = client.list_queues(parent)
+
+    # Print the results.
+    for queue in response:
+        print(queue.name)
+
+    if response.num_results == 0:
+        print('No queues found!')
+# [END cloud_tasks_list_queues]

tasks/list_queues_test.py

@@ -0,0 +1,55 @@
+# Copyright 2020 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.
+
+import os
+import uuid
+
+from google.cloud import tasks_v2
+import pytest
+
+import list_queues
+
+TEST_PROJECT_ID = os.environ['GOOGLE_CLOUD_PROJECT']
+TEST_LOCATION = os.getenv('TEST_QUEUE_LOCATION', 'us-central1')
+TEST_QUEUE_NAME = f'my-queue-{uuid.uuid4().hex}'
+
+
+@pytest.fixture()
+def test_queue():
+    client = tasks_v2.CloudTasksClient()
+    parent = client.location_path(TEST_PROJECT_ID, TEST_LOCATION)
+    queue = {
+        # The fully qualified path to the queue
+        'name': client.queue_path(
+            TEST_PROJECT_ID, TEST_LOCATION, TEST_QUEUE_NAME),
+    }
+    q = client.create_queue(parent, queue)
+
+    yield q
+
+    client.delete_queue(q.name)
+
+
+def test_list_queues_not_present(capsys):
+    list_queues.list_queues(TEST_PROJECT_ID, TEST_LOCATION)
+    out, _ = capsys.readouterr()
+
+    assert(TEST_QUEUE_NAME not in out)
+
+
+def test_list_queues_present(capsys, test_queue):
+    list_queues.list_queues(TEST_PROJECT_ID, TEST_LOCATION)
+    out, _ = capsys.readouterr()
+
+    assert(TEST_QUEUE_NAME in out)