samples: add async upload sample (#665)

* samples: add async upload sample

* make python3.6 compatible

* woops, this one too

* so annoying, can we deprecate 3.6 too?

* be gone with you 3.6!

* add comment clarifying how sample is run

samples/snippets/snippets_test.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import os
 import tempfile
 import time
@@ -23,6 +24,7 @@
 import requests
 
 import storage_add_bucket_label
+import storage_async_upload
 import storage_batch_request
 import storage_bucket_delete_default_kms_key
 import storage_change_default_storage_class
@@ -213,6 +215,12 @@ def test_upload_blob_with_kms(test_bucket):
         assert kms_blob.kms_key_name.startswith(KMS_KEY)
 
 
+def test_async_upload(bucket, capsys):
+    asyncio.run(storage_async_upload.async_upload_blob(bucket.name))
+    out, _ = capsys.readouterr()
+    assert f"Uploaded 3 files to bucket {bucket.name}" in out
+
+
 def test_download_byte_range(test_blob):
     with tempfile.NamedTemporaryFile() as dest_file:
         storage_download_byte_range.download_byte_range(

samples/snippets/storage_async_upload.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+
+# Copyright 2021 Google Inc. 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.
+
+import asyncio
+import sys
+
+
+"""Sample that asynchronously uploads a file to GCS
+"""
+
+
+# [START storage_async_upload]
+# This sample can be run by calling `async.run(async_upload_blob('bucket_name'))`
+async def async_upload_blob(bucket_name):
+    """Uploads a number of files in parallel to the bucket."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+    import asyncio
+    from functools import partial
+    from google.cloud import storage
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    loop = asyncio.get_running_loop()
+
+    tasks = []
+    count = 3
+    for x in range(count):
+        blob_name = f"async_sample_blob_{x}"
+        content = f"Hello world #{x}"
+        blob = bucket.blob(blob_name)
+        # The first arg, None, tells it to use the default loops executor
+        tasks.append(loop.run_in_executor(None, partial(blob.upload_from_string, content)))
+
+    # If the method returns a value (such as download_as_string), gather will return the values
+    await asyncio.gather(*tasks)
+
+    print(f"Uploaded {count} files to bucket {bucket_name}")
+
+
+# [END storage_async_upload]
+
+
+if __name__ == "__main__":
+    asyncio.run(async_upload_blob(
+        bucket_name=sys.argv[1]
+    ))