Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bautsi wants to merge 2 commits into
base: main
Choose a base branch
from
+55 −47
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4450786
chore(storage): improve type annotations in client and bucket modules
bautsi Apr 13, 2026
dd55f62
fix(storage): resolve circular import issues in type hints
bautsi Apr 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 32 additions & 25 deletions packages/google-cloud-storage/google/cloud/storage/bucket.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,14 @@
import datetime
import json
import warnings
from typing import Any, Optional, Set, Tuple, Union, TYPE_CHECKING
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._helpers import (
_NOW,
Expand Down Expand Up @@ -65,8 +66,12 @@
DEFAULT_RETRY_IF_ETAG_IN_JSON,
DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
ConditionalRetryPolicy,
)

if TYPE_CHECKING:
from google.cloud.storage.client import Client

_UBLA_BPO_ENABLED_MESSAGE = (
"Pass only one of 'uniform_bucket_level_access_enabled' / "
"'bucket_policy_only_enabled' to 'IAMConfiguration'."
Expand Down Expand Up @@ -846,12 +851,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::
Expand All @@ -862,9 +867,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:
Expand Down Expand Up @@ -1289,21 +1296,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)
Expand Down
45 changes: 23 additions & 22 deletions packages/google-cloud-storage/google/cloud/storage/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,15 +22,16 @@
import json
import os
import warnings
from typing import Optional, Sequence, Tuple, Union

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
from google.cloud.client import ClientWithProject
from google.cloud.exceptions import NotFound

from google.cloud.storage._helpers import (
_DEFAULT_SCHEME,
_DEFAULT_UNIVERSE_DOMAIN,
Expand Down Expand Up @@ -58,7 +59,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()

Expand Down Expand Up @@ -881,15 +882,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).
Expand Down Expand Up @@ -1012,18 +1013,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).
Expand Down
Loading