chore(storage): improve type annotations in client and bucket modules

Author: bautsi

packages/google-cloud-storage/google/cloud/storage/bucket.py

@@ -18,15 +18,17 @@
 import copy
 import datetime
 import json
+from typing import Any, Optional, Set, Tuple, Union
 import warnings
 from urllib.parse import urlsplit
 
 from google.api_core import datetime_helpers
 from google.api_core.iam import Policy
+from google.api_core.retry import Retry
 from google.cloud._helpers import _datetime_to_rfc3339, _rfc3339_nanos_to_datetime
 from google.cloud.exceptions import NotFound
 
-from google.cloud.storage import _signing
+from google.cloud.storage import Client, _signing
 from google.cloud.storage._helpers import (
     _NOW,
     _UTC,
@@ -65,6 +67,7 @@
     DEFAULT_RETRY_IF_ETAG_IN_JSON,
     DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
+    ConditionalRetryPolicy,
 )
 
 _UBLA_BPO_ENABLED_MESSAGE = (
@@ -846,12 +849,12 @@ def from_string(cls, uri, client=None):
 
     def blob(
         self,
-        blob_name,
-        chunk_size=None,
-        encryption_key=None,
-        kms_key_name=None,
-        generation=None,
-    ):
+        blob_name: str,
+        chunk_size: Optional[int] = None,
+        encryption_key: Optional[bytes] = None,
+        kms_key_name: Optional[str] = None,
+        generation: Optional[int] = None,
+    ) -> Blob:
         """Factory constructor for blob object.
 
         .. note::
@@ -862,9 +865,11 @@ def blob(
         :param blob_name: The name of the blob to be instantiated.
 
         :type chunk_size: int
-        :param chunk_size: The size of a chunk of data whenever iterating
-                           (in bytes). This must be a multiple of 256 KB per
-                           the API specification.
+        :param chunk_size:
+            (Optional) The size of a chunk of data whenever iterating (in bytes).
+            This must be a multiple of 256 KB per the API specification. If not
+            specified, the chunk_size of the blob itself is used. If that is not
+            specified, a default value of 40 MB is used.
 
         :type encryption_key: bytes
         :param encryption_key:
@@ -1289,21 +1294,21 @@ def path(self):
 
     def get_blob(
         self,
-        blob_name,
-        client=None,
-        encryption_key=None,
-        generation=None,
-        if_etag_match=None,
-        if_etag_not_match=None,
-        if_generation_match=None,
-        if_generation_not_match=None,
-        if_metageneration_match=None,
-        if_metageneration_not_match=None,
-        timeout=_DEFAULT_TIMEOUT,
-        retry=DEFAULT_RETRY,
-        soft_deleted=None,
-        **kwargs,
-    ):
+        blob_name: str,
+        client: Optional[Client] = None,
+        encryption_key: Optional[bytes] = None,
+        generation: Optional[int] = None,
+        if_etag_match: Optional[Union[str, Set[str]]] = None,
+        if_etag_not_match: Optional[Union[str, Set[str]]] = None,
+        if_generation_match: Optional[int] = None,
+        if_generation_not_match: Optional[int] = None,
+        if_metageneration_match: Optional[int] = None,
+        if_metageneration_not_match: Optional[int] = None,
+        timeout: Optional[Union[float, Tuple[float, float]]] = _DEFAULT_TIMEOUT,
+        retry: Optional[Union[Retry, ConditionalRetryPolicy]] = DEFAULT_RETRY,
+        soft_deleted: Optional[bool] = None,
+        **kwargs: Any,
+    ) -> Optional[Blob]:
         """Get a blob object by name.
 
         See a [code sample](https://cloud.google.com/storage/docs/samples/storage-get-metadata#storage_get_metadata-python)

packages/google-cloud-storage/google/cloud/storage/client.py

@@ -21,10 +21,12 @@
 import functools
 import json
 import os
+from typing import Optional, Sequence, Tuple, Union
 import warnings
 
 import google.api_core.client_options
 from google.api_core import page_iterator
+from google.api_core.retry import Retry
 from google.auth.credentials import AnonymousCredentials
 from google.auth.transport import mtls
 from google.cloud._helpers import _LocalStack
@@ -58,7 +60,7 @@
 from google.cloud.storage.bucket import Bucket, _blobs_page_start, _item_to_blob
 from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 from google.cloud.storage.hmac_key import HMACKeyMetadata
-from google.cloud.storage.retry import DEFAULT_RETRY
+from google.cloud.storage.retry import DEFAULT_RETRY, ConditionalRetryPolicy
 
 _marker = object()
 
@@ -881,15 +883,15 @@ def _bucket_arg_to_bucket(self, bucket_or_name, generation=None):
 
     def get_bucket(
         self,
-        bucket_or_name,
-        timeout=_DEFAULT_TIMEOUT,
-        if_metageneration_match=None,
-        if_metageneration_not_match=None,
-        retry=DEFAULT_RETRY,
+        bucket_or_name: Union[Bucket, str],
+        timeout: Optional[Union[float, Tuple[float, float]]] = _DEFAULT_TIMEOUT,
+        if_metageneration_match: Optional[int] = None,
+        if_metageneration_not_match: Optional[int] = None,
+        retry: Optional[Union[Retry, ConditionalRetryPolicy]] = DEFAULT_RETRY,
         *,
-        generation=None,
-        soft_deleted=None,
-    ):
+        generation: Optional[int] = None,
+        soft_deleted: Optional[bool] = None,
+    ) -> Bucket:
         """Retrieve a bucket via a GET request.
 
         See [API reference docs](https://cloud.google.com/storage/docs/json_api/v1/buckets/get) and a [code sample](https://cloud.google.com/storage/docs/samples/storage-get-bucket-metadata#storage_get_bucket_metadata-python).
@@ -1012,18 +1014,18 @@ def lookup_bucket(
 
     def create_bucket(
         self,
-        bucket_or_name,
-        requester_pays=None,
-        project=None,
-        user_project=None,
-        location=None,
-        data_locations=None,
-        predefined_acl=None,
-        predefined_default_object_acl=None,
-        enable_object_retention=False,
-        timeout=_DEFAULT_TIMEOUT,
-        retry=DEFAULT_RETRY,
-    ):
+        bucket_or_name: Union[Bucket, str],
+        requester_pays: Optional[bool] = None,
+        project: Optional[str] = None,
+        user_project: Optional[str] = None,
+        location: Optional[str] = None,
+        data_locations: Optional[Sequence[str]] = None,
+        predefined_acl: Optional[str] = None,
+        predefined_default_object_acl: Optional[str] = None,
+        enable_object_retention: Optional[bool] = False,
+        timeout: Optional[Union[float, Tuple[float, float]]] = _DEFAULT_TIMEOUT,
+        retry: Optional[Union[Retry, ConditionalRetryPolicy]] = DEFAULT_RETRY,
+    ) -> Bucket:
         """Create a new bucket via a POST request.
 
         See [API reference docs](https://cloud.google.com/storage/docs/json_api/v1/buckets/insert) and a [code sample](https://cloud.google.com/storage/docs/samples/storage-create-bucket#storage_create_bucket-python).