Type hints for s3 storage sdk (GoogleCloudPlatform#9947)

askmeegs · engelke · web-flow · commit dd0f1b85b02f · 2023-05-18T08:59:33.000-07:00
* Type hunts for s3 storage sdk

* attempt to fix linter

* type hint return values

* fix return type hint

* Failing tests

* fix: minor type tweaks

* fix: import typing for older Python versions

* Update list_gcs_buckets.py

* Update list_gcs_objects.py

* Update s3_sdk_test.py

* fix lint

* lint fixes

* fix type

---------

Co-authored-by: Charles Engelke &lt;engelke@google.com&gt;
diff --git a/storage/s3-sdk/list_gcs_buckets.py b/storage/s3-sdk/list_gcs_buckets.py
@@ -15,12 +15,15 @@
 # limitations under the License.
 
 import argparse
+from typing import List
 
 # [START storage_s3_sdk_list_buckets]
-import boto3
+import boto3  # type: ignore
 
 
-def list_gcs_buckets(google_access_key_id, google_access_key_secret):
+def list_gcs_buckets(
+    google_access_key_id: str, google_access_key_secret: str
+) -> List[str]:
     """Lists all Cloud Storage buckets using AWS SDK for Python (boto3)
     Positional arguments:
         google_access_key_id: hash-based message authentication code (HMAC) access ID
diff --git a/storage/s3-sdk/list_gcs_objects.py b/storage/s3-sdk/list_gcs_objects.py
@@ -15,12 +15,15 @@
 # limitations under the License.
 
 import argparse
+from typing import List
 
 # [START storage_s3_sdk_list_objects]
-import boto3
+import boto3  # type: ignore
 
 
-def list_gcs_objects(google_access_key_id, google_access_key_secret, bucket_name):
+def list_gcs_objects(
+    google_access_key_id: str, google_access_key_secret: str, bucket_name: str
+) -> List[str]:
     """Lists all Cloud Storage objects using AWS SDK for Python (boto3)
     Positional arguments:
         google_access_key_id: hash-based message authentication code (HMAC) access ID
diff --git a/storage/s3-sdk/noxfile_config.py b/storage/s3-sdk/noxfile_config.py
@@ -21,7 +21,7 @@
 # RUN_TESTS_SESSION. The reason we can not use multiple project is
 # that our new projects are enforced to have
 # 'constraints/iam.disableServiceAccountKeyCreation' policy.
-def get_service_account_email():
+def get_service_account_email() -> str:
     session = os.environ.get("RUN_TESTS_SESSION")
     if session == "py-3.6":
         return "py36-storage-test@" "python-docs-samples-tests.iam.gserviceaccount.com"
diff --git a/storage/s3-sdk/s3_sdk_test.py b/storage/s3-sdk/s3_sdk_test.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import os
+from typing import Tuple
 import uuid
 
 import backoff
@@ -30,7 +31,7 @@
 
 
 @pytest.fixture(scope="module")
-def hmac_fixture():
+def hmac_fixture() -> Tuple[storage.hmac_key.HMACKeyMetadata, str]:
     """
     Creates an HMAC Key and secret to supply to the S3 SDK tests. The key
     will be deleted after the test session.
@@ -45,7 +46,7 @@ def hmac_fixture():
 
 
 @pytest.fixture(scope="module")
-def test_bucket():
+def test_bucket() -> storage.Bucket:
     """Yields a bucket that is deleted after the test completes."""
     bucket = None
     while bucket is None or bucket.exists():
@@ -57,7 +58,7 @@ def test_bucket():
 
 
 @pytest.fixture(scope="module")
-def test_blob(test_bucket):
+def test_blob(test_bucket: storage.Bucket) -> storage.Blob:
     """Yields a blob that is deleted after the test completes."""
     bucket = test_bucket
     blob = bucket.blob(f"storage_snippets_test_sigil-{uuid.uuid4()}")
@@ -68,18 +69,20 @@ def test_blob(test_bucket):
 # Retry request because the created key may not be fully propagated for up
 # to 15s.
 @backoff.on_exception(backoff.constant, ClientError, interval=1, max_time=15)
-def test_list_buckets(hmac_fixture, test_bucket):
-    results = list_gcs_buckets.list_gcs_buckets(
+def test_list_buckets(hmac_fixture: Tuple[storage.hmac_key.HMACKeyMetadata, str], test_bucket: storage.Bucket) -> None:
+    result = list_gcs_buckets.list_gcs_buckets(
         google_access_key_id=hmac_fixture[0].access_id,
         google_access_key_secret=hmac_fixture[1],
     )
-    assert test_bucket.name in results
+    assert test_bucket.name in result
 
 
 # Retry request because the created key may not be fully propagated for up
 # to 15s.
 @backoff.on_exception(backoff.constant, ClientError, interval=1, max_time=15)
-def test_list_blobs(hmac_fixture, test_bucket, test_blob):
+def test_list_blobs(
+    hmac_fixture: Tuple[storage.hmac_key.HMACKeyMetadata, str], test_bucket: storage.Bucket, test_blob: storage.Blob
+) -> None:
     result = list_gcs_objects.list_gcs_objects(
         google_access_key_id=hmac_fixture[0].access_id,
         google_access_key_secret=hmac_fixture[1],
