# Copyright 2015 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.

"""Client for interacting with the Google Cloud Storage API."""

import base64

import binascii

import collections

import datetime

import functools

import json

import os

import warnings

import google.api_core.client_options

from google.auth.credentials import AnonymousCredentials

from google.auth.transport import mtls

from google.api_core import page_iterator

from google.cloud._helpers import _LocalStack

from google.cloud.client import ClientWithProject

from google.cloud.exceptions import NotFound

from google.cloud.storage._helpers import _add_generation_match_parameters

from google.cloud.storage._helpers import _bucket_bound_hostname_url

from google.cloud.storage._helpers import _get_api_endpoint_override

from google.cloud.storage._helpers import _get_environ_project

from google.cloud.storage._helpers import _get_storage_emulator_override

from google.cloud.storage._helpers import _virtual_hosted_style_base_url

from google.cloud.storage._helpers import _DEFAULT_UNIVERSE_DOMAIN

from google.cloud.storage._helpers import _DEFAULT_SCHEME

from google.cloud.storage._helpers import _STORAGE_HOST_TEMPLATE

from google.cloud.storage._helpers import _NOW

from google.cloud.storage._helpers import _UTC

from google.cloud.storage._opentelemetry_tracing import create_trace_span

from google.cloud.storage._http import Connection

from google.cloud.storage._signing import (

get_expiration_seconds_v4,

get_v4_now_dtstamps,

ensure_signed_credentials,

_sign_message,

)

from google.cloud.storage.batch import Batch

from google.cloud.storage.bucket import Bucket, _item_to_blob, _blobs_page_start

from google.cloud.storage.blob import Blob

from google.cloud.storage.hmac_key import HMACKeyMetadata

from google.cloud.storage.acl import BucketACL

from google.cloud.storage.acl import DefaultObjectACL

from google.cloud.storage.constants import _DEFAULT_TIMEOUT

from google.cloud.storage.retry import DEFAULT_RETRY

_marker = object()

def _buckets_page_start(iterator, page, response):

"""Grab unreachable buckets after a :class:`~google.cloud.iterator.Page` started."""

unreachable = response.get("unreachable", [])

if not isinstance(unreachable, list):

raise TypeError(

f"expected unreachable to be list, but obtained {type(unreachable)}"

)

page.unreachable = unreachable

class Client(ClientWithProject):

"""Client to bundle configuration needed for API requests.

:type project: str or None

:param project: the project which the client acts on behalf of. Will be

passed when creating a topic. If not passed,

falls back to the default inferred from the environment.

:type credentials: :class:`~google.auth.credentials.Credentials`

:param credentials: (Optional) The OAuth2 Credentials to use for this

client. If not passed (and if no ``_http`` object is

passed), falls back to the default inferred from the

environment.

:type _http: :class:`~requests.Session`

:param _http: (Optional) HTTP object to make requests. Can be any object

that defines ``request()`` with the same interface as

:meth:`requests.Session.request`. If not passed, an

``_http`` object is created that is bound to the

``credentials`` for the current object.

This parameter should be considered private, and could

change in the future.

:type client_info: :class:`~google.api_core.client_info.ClientInfo`

:param client_info:

The client info used to send a user-agent string along with API

requests. If ``None``, then default info will be used. Generally,

you only need to set this if you're developing your own library

or partner tool.

:type client_options: :class:`~google.api_core.client_options.ClientOptions` or :class:`dict`

:param client_options: (Optional) Client options used to set user options on the client.

A non-default universe domain or api endpoint should be set through client_options.

:type use_auth_w_custom_endpoint: bool

:param use_auth_w_custom_endpoint:

(Optional) Whether authentication is required under custom endpoints.

If false, uses AnonymousCredentials and bypasses authentication.

Defaults to True. Note this is only used when a custom endpoint is set in conjunction.

:type extra_headers: dict

:param extra_headers:

(Optional) Custom headers to be sent with the requests attached to the client.

For example, you can add custom audit logging headers.

:type api_key: string

:param api_key:

(Optional) An API key. Mutually exclusive with any other credentials.

This parameter is an alias for setting `client_options.api_key` and

will supercede any api key set in the `client_options` parameter.

"""

SCOPE = (

"https://www.googleapis.com/auth/devstorage.full_control",

"https://www.googleapis.com/auth/devstorage.read_only",

"https://www.googleapis.com/auth/devstorage.read_write",

)

"""The scopes required for authenticating as a Cloud Storage consumer."""

def __init__(

self,

project=_marker,

credentials=None,

_http=None,

client_info=None,

client_options=None,

use_auth_w_custom_endpoint=True,

extra_headers={},

*,

api_key=None,

):

self._base_connection = None

if project is None:

no_project = True

project = "<none>"

else:

no_project = False

if project is _marker:

project = None

# Save the initial value of constructor arguments before they

# are passed along, for use in __reduce__ defined elsewhere.

self._initial_client_info = client_info

self._initial_client_options = client_options

self._extra_headers = extra_headers

connection_kw_args = {"client_info": client_info}

# api_key should set client_options.api_key. Set it here whether

# client_options was specified as a dict, as a ClientOptions object, or

# None.

if api_key:

if client_options and not isinstance(client_options, dict):

client_options.api_key = api_key

else:

if not client_options:

client_options = {}

client_options["api_key"] = api_key

if client_options:

if isinstance(client_options, dict):

client_options = google.api_core.client_options.from_dict(

client_options

)

if client_options and client_options.universe_domain:

self._universe_domain = client_options.universe_domain

else:

self._universe_domain = None

storage_emulator_override = _get_storage_emulator_override()

api_endpoint_override = _get_api_endpoint_override()

# Determine the api endpoint. The rules are as follows:

# 1. If the `api_endpoint` is set in `client_options`, use that as the

# endpoint.

if client_options and client_options.api_endpoint:

api_endpoint = client_options.api_endpoint

# 2. Elif the "STORAGE_EMULATOR_HOST" env var is set, then use that as the

# endpoint.

elif storage_emulator_override:

api_endpoint = storage_emulator_override

# 3. Elif the "API_ENDPOINT_OVERRIDE" env var is set, then use that as the

# endpoint.

elif api_endpoint_override:

api_endpoint = api_endpoint_override

# 4. Elif the `universe_domain` is set in `client_options`,

# create the endpoint using that as the default.

#

# Mutual TLS is not compatible with a non-default universe domain

# at this time. If such settings are enabled along with the

# "GOOGLE_API_USE_CLIENT_CERTIFICATE" env variable, a ValueError will

# be raised.

elif self._universe_domain:

# The final decision of whether to use mTLS takes place in

# google-auth-library-python. We peek at the environment variable

# here only to issue an exception in case of a conflict.

use_client_cert = False

if hasattr(mtls, "should_use_client_cert"):

use_client_cert = mtls.should_use_client_cert()

else:

use_client_cert = (

os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE") == "true"

)

if use_client_cert:

raise ValueError(

'The "GOOGLE_API_USE_CLIENT_CERTIFICATE" env variable is '

'set to "true" and a non-default universe domain is '

"configured. mTLS is not supported in any universe other than"

"googleapis.com."

)

api_endpoint = _DEFAULT_SCHEME + _STORAGE_HOST_TEMPLATE.format(

universe_domain=self._universe_domain

)

# 5. Else, use the default, which is to use the default

# universe domain of "googleapis.com" and create the endpoint

# "storage.googleapis.com" from that.

else:

api_endpoint = None

connection_kw_args["api_endpoint"] = api_endpoint

self._is_emulator_set = True if storage_emulator_override else False

# If a custom endpoint is set, the client checks for credentials

# or finds the default credentials based on the current environment.

# Authentication may be bypassed under certain conditions:

# (1) STORAGE_EMULATOR_HOST is set (for backwards compatibility), OR

# (2) use_auth_w_custom_endpoint is set to False.

if connection_kw_args["api_endpoint"] is not None:

if self._is_emulator_set or not use_auth_w_custom_endpoint:

if credentials is None:

credentials = AnonymousCredentials()

if project is None:

project = _get_environ_project()

if project is None:

no_project = True

project = "<none>"

super(Client, self).__init__(

project=project,

credentials=credentials,

client_options=client_options,

_http=_http,

)

# Validate that the universe domain of the credentials matches the

# universe domain of the client.

if self._credentials.universe_domain != self.universe_domain:

raise ValueError(

"The configured universe domain ({client_ud}) does not match "

"the universe domain found in the credentials ({cred_ud}). If "

"you haven't configured the universe domain explicitly, "

"`googleapis.com` is the default.".format(

client_ud=self.universe_domain,

cred_ud=self._credentials.universe_domain,

)

)

if no_project:

self.project = None

# Pass extra_headers to Connection

connection = Connection(self, **connection_kw_args)

connection.extra_headers = extra_headers

self._connection = connection

self._batch_stack = _LocalStack()

@classmethod

def create_anonymous_client(cls):

"""Factory: return client with anonymous credentials.

.. note::

Such a client has only limited access to "public" buckets:

listing their contents and downloading their blobs.

:rtype: :class:`google.cloud.storage.client.Client`

:returns: Instance w/ anonymous credentials and no project.

"""

client = cls(project="<none>", credentials=AnonymousCredentials())

client.project = None

return client

@property

def universe_domain(self):

return self._universe_domain or _DEFAULT_UNIVERSE_DOMAIN

@property

def api_endpoint(self):

return self._connection.API_BASE_URL

def update_user_agent(self, user_agent):

"""Update the user-agent string for this client.

:type user_agent: str

:param user_agent: The string to add to the user-agent.

"""

existing_user_agent = self._connection._client_info.user_agent

if existing_user_agent is None:

self._connection.user_agent = user_agent

else:

self._connection.user_agent = f"{user_agent} {existing_user_agent}"

@property

def _connection(self):

"""Get connection or batch on the client.

:rtype: :class:`google.cloud.storage._http.Connection`

:returns: The connection set on the client, or the batch

if one is set.

"""

if self.current_batch is not None:

return self.current_batch

else:

return self._base_connection

@_connection.setter

def _connection(self, value):

"""Set connection on the client.

Intended to be used by constructor (since the base class calls)

self._connection = connection

Will raise if the connection is set more than once.

:type value: :class:`google.cloud.storage._http.Connection`

:param value: The connection set on the client.

:raises: :class:`ValueError` if connection has already been set.

"""

if self._base_connection is not None:

raise ValueError("Connection already set on client")

self._base_connection = value

def _push_batch(self, batch):

"""Push a batch onto our stack.

"Protected", intended for use by batch context mgrs.

:type batch: :class:`google.cloud.storage.batch.Batch`

:param batch: newly-active batch

"""

self._batch_stack.push(batch)

def _pop_batch(self):

"""Pop a batch from our stack.

"Protected", intended for use by batch context mgrs.

:raises: IndexError if the stack is empty.

:rtype: :class:`google.cloud.storage.batch.Batch`

:returns: the top-most batch/transaction, after removing it.

"""

return self._batch_stack.pop()

@property

def current_batch(self):

"""Currently-active batch.

:rtype: :class:`google.cloud.storage.batch.Batch` or ``NoneType`` (if

no batch is active).

:returns: The batch at the top of the batch stack.

"""

return self._batch_stack.top

def get_service_account_email(

self, project=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY

):

"""Get the email address of the project's GCS service account

:type project: str

:param project:

(Optional) Project ID to use for retreiving GCS service account

email address. Defaults to the client's project.

:type timeout: float or tuple

:param timeout:

(Optional) The amount of time, in seconds, to wait

for the server response. See: :ref:`configuring_timeouts`

:type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy

:param retry:

(Optional) How to retry the RPC. See: :ref:`configuring_retries`

:rtype: str

:returns: service account email address

"""

with create_trace_span(name="Storage.Client.getServiceAccountEmail"):

if project is None:

project = self.project

path = f"/projects/{project}/serviceAccount"

api_response = self._get_resource(path, timeout=timeout, retry=retry)

return api_response["email_address"]

def bucket(self, bucket_name, user_project=None, generation=None):

"""Factory constructor for bucket object.

.. note::

This will not make an HTTP request; it simply instantiates

a bucket object owned by this client.

:type bucket_name: str

:param bucket_name: The name of the bucket to be instantiated.

:type user_project: str

:param user_project: (Optional) The project ID to be billed for API

requests made via the bucket.

:type generation: int

:param generation: (Optional) If present, selects a specific revision of

this bucket.

:rtype: :class:`google.cloud.storage.bucket.Bucket`

:returns: The bucket object created.

"""

return Bucket(

client=self,

name=bucket_name,

user_project=user_project,

generation=generation,

)

def batch(self, raise_exception=True):

"""Factory constructor for batch object.

.. note::

This will not make an HTTP request; it simply instantiates

a batch object owned by this client.

:type raise_exception: bool

:param raise_exception:

(Optional) Defaults to True. If True, instead of adding exceptions

to the list of return responses, the final exception will be raised.

Note that exceptions are unwrapped after all operations are complete

in success or failure, and only the last exception is raised.

:rtype: :class:`google.cloud.storage.batch.Batch`

:returns: The batch object created.

"""

return Batch(client=self, raise_exception=raise_exception)

def _get_resource(

self,

path,

query_params=None,

headers=None,

timeout=_DEFAULT_TIMEOUT,

retry=DEFAULT_RETRY,

_target_object=None,

):

"""Helper for bucket / blob methods making API 'GET' calls.

Args:

path str:

The path of the resource to fetch.

query_params Optional[dict]:

HTTP query parameters to be passed

headers Optional[dict]:

HTTP headers to be passed

timeout (Optional[Union[float, Tuple[float, float]]]):

The amount of time, in seconds, to wait for the server response.

Can also be passed as a tuple (connect_timeout, read_timeout).

See :meth:`requests.Session.request` documentation for details.

retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):

How to retry the RPC. A None value will disable retries.

A google.api_core.retry.Retry value will enable retries, and the object will

define retriable response codes and errors and configure backoff and timeout options.

A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and

activates it only if certain conditions are met. This class exists to provide safe defaults

for RPC calls that are not technically safe to retry normally (due to potential data

duplication or other side-effects) but become safe to retry if a condition such as

if_metageneration_match is set.

See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for

information on retry types and how to configure them.

_target_object (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

:class:`~google.cloud.storage.bucket.blob`, \

]):

Object to which future data is to be applied -- only relevant

in the context of a batch.

Returns:

dict

The JSON resource fetched

Raises:

google.cloud.exceptions.NotFound

If the bucket is not found.

"""

return self._connection.api_request(

method="GET",

path=path,

query_params=query_params,

headers=headers,

timeout=timeout,

retry=retry,

_target_object=_target_object,

)

def _list_resource(

self,

path,

item_to_value,

page_token=None,

max_results=None,

extra_params=None,

page_start=page_iterator._do_nothing_page_start,

page_size=None,

timeout=_DEFAULT_TIMEOUT,

retry=DEFAULT_RETRY,

):

kwargs = {

"method": "GET",

"path": path,

"timeout": timeout,

}

with create_trace_span(

name="Storage.Client._list_resource_returns_iterator",

client=self,

api_request=kwargs,

retry=retry,

):

api_request = functools.partial(

self._connection.api_request, timeout=timeout, retry=retry

)

return page_iterator.HTTPIterator(

client=self,

api_request=api_request,

path=path,

item_to_value=item_to_value,

page_token=page_token,

max_results=max_results,

extra_params=extra_params,

page_start=page_start,

page_size=page_size,

)

def _patch_resource(

self,

path,

data,

query_params=None,

headers=None,

timeout=_DEFAULT_TIMEOUT,

retry=None,

_target_object=None,

):

"""Helper for bucket / blob methods making API 'PATCH' calls.

Args:

path str:

The path of the resource to fetch.

data dict:

The data to be patched.

query_params Optional[dict]:

HTTP query parameters to be passed

headers Optional[dict]:

HTTP headers to be passed

timeout (Optional[Union[float, Tuple[float, float]]]):

The amount of time, in seconds, to wait for the server response.

Can also be passed as a tuple (connect_timeout, read_timeout).

See :meth:`requests.Session.request` documentation for details.

retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):

How to retry the RPC. A None value will disable retries.

A google.api_core.retry.Retry value will enable retries, and the object will

define retriable response codes and errors and configure backoff and timeout options.

A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and

activates it only if certain conditions are met. This class exists to provide safe defaults

for RPC calls that are not technically safe to retry normally (due to potential data

duplication or other side-effects) but become safe to retry if a condition such as

if_metageneration_match is set.

See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for

information on retry types and how to configure them.

_target_object (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

:class:`~google.cloud.storage.bucket.blob`, \

]):

Object to which future data is to be applied -- only relevant

in the context of a batch.

Returns:

dict

The JSON resource fetched

Raises:

google.cloud.exceptions.NotFound

If the bucket is not found.

"""

return self._connection.api_request(

method="PATCH",

path=path,

data=data,

query_params=query_params,

headers=headers,

timeout=timeout,

retry=retry,

_target_object=_target_object,

)

def _put_resource(

self,

path,

data,

query_params=None,

headers=None,

timeout=_DEFAULT_TIMEOUT,

retry=None,

_target_object=None,

):

"""Helper for bucket / blob methods making API 'PUT' calls.

Args:

path str:

The path of the resource to fetch.

data dict:

The data to be patched.

query_params Optional[dict]:

HTTP query parameters to be passed

headers Optional[dict]:

HTTP headers to be passed

timeout (Optional[Union[float, Tuple[float, float]]]):

The amount of time, in seconds, to wait for the server response.

Can also be passed as a tuple (connect_timeout, read_timeout).

See :meth:`requests.Session.request` documentation for details.

retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):

How to retry the RPC. A None value will disable retries.

A google.api_core.retry.Retry value will enable retries, and the object will

define retriable response codes and errors and configure backoff and timeout options.

A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and

activates it only if certain conditions are met. This class exists to provide safe defaults

for RPC calls that are not technically safe to retry normally (due to potential data

duplication or other side-effects) but become safe to retry if a condition such as

if_metageneration_match is set.

See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for

information on retry types and how to configure them.

_target_object (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

:class:`~google.cloud.storage.bucket.blob`, \

]):

Object to which future data is to be applied -- only relevant

in the context of a batch.

Returns:

dict

The JSON resource fetched

Raises:

google.cloud.exceptions.NotFound

If the bucket is not found.

"""

return self._connection.api_request(

method="PUT",

path=path,

data=data,

query_params=query_params,

headers=headers,

timeout=timeout,

retry=retry,

_target_object=_target_object,

)

def _post_resource(

self,

path,

data,

query_params=None,

headers=None,

timeout=_DEFAULT_TIMEOUT,

retry=None,

_target_object=None,

):

"""Helper for bucket / blob methods making API 'POST' calls.

Args:

path str:

The path of the resource to which to post.

data dict:

The data to be posted.

query_params Optional[dict]:

HTTP query parameters to be passed

headers Optional[dict]:

HTTP headers to be passed

timeout (Optional[Union[float, Tuple[float, float]]]):

The amount of time, in seconds, to wait for the server response.

Can also be passed as a tuple (connect_timeout, read_timeout).

See :meth:`requests.Session.request` documentation for details.

retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):

How to retry the RPC. A None value will disable retries.

A google.api_core.retry.Retry value will enable retries, and the object will

define retriable response codes and errors and configure backoff and timeout options.

A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and

activates it only if certain conditions are met. This class exists to provide safe defaults

for RPC calls that are not technically safe to retry normally (due to potential data

duplication or other side-effects) but become safe to retry if a condition such as

if_metageneration_match is set.

See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for

information on retry types and how to configure them.

_target_object (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

:class:`~google.cloud.storage.bucket.blob`, \

]):

Object to which future data is to be applied -- only relevant

in the context of a batch.

Returns:

dict

The JSON resource returned from the post.

Raises:

google.cloud.exceptions.NotFound

If the bucket is not found.

"""

return self._connection.api_request(

method="POST",

path=path,

data=data,

query_params=query_params,

headers=headers,

timeout=timeout,

retry=retry,

_target_object=_target_object,

)

def _delete_resource(

self,

path,

query_params=None,

headers=None,

timeout=_DEFAULT_TIMEOUT,

retry=DEFAULT_RETRY,

_target_object=None,

):

"""Helper for bucket / blob methods making API 'DELETE' calls.

Args:

path str:

The path of the resource to delete.

query_params Optional[dict]:

HTTP query parameters to be passed

headers Optional[dict]:

HTTP headers to be passed

timeout (Optional[Union[float, Tuple[float, float]]]):

The amount of time, in seconds, to wait for the server response.

Can also be passed as a tuple (connect_timeout, read_timeout).

See :meth:`requests.Session.request` documentation for details.

retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):

How to retry the RPC. A None value will disable retries.

A google.api_core.retry.Retry value will enable retries, and the object will

define retriable response codes and errors and configure backoff and timeout options.

A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and

activates it only if certain conditions are met. This class exists to provide safe defaults

for RPC calls that are not technically safe to retry normally (due to potential data

duplication or other side-effects) but become safe to retry if a condition such as

if_metageneration_match is set.

See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for

information on retry types and how to configure them.

_target_object (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

:class:`~google.cloud.storage.bucket.blob`, \

]):

Object to which future data is to be applied -- only relevant

in the context of a batch.

Returns:

dict

The JSON resource fetched

Raises:

google.cloud.exceptions.NotFound

If the bucket is not found.

"""

return self._connection.api_request(

method="DELETE",

path=path,

query_params=query_params,

headers=headers,

timeout=timeout,

retry=retry,

_target_object=_target_object,

)

def _bucket_arg_to_bucket(self, bucket_or_name, generation=None):

"""Helper to return given bucket or create new by name.

Args:

bucket_or_name (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

str, \

]):

The bucket resource to pass or name to create.

generation (Optional[int]):

The bucket generation. If generation is specified,

bucket_or_name must be a name (str).

Returns:

google.cloud.storage.bucket.Bucket

The newly created bucket or the given one.

"""

if isinstance(bucket_or_name, Bucket):

if generation:

raise ValueError(

"The generation can only be specified if a "

"name is used to specify a bucket, not a Bucket object. "

"Create a new Bucket object with the correct generation "

"instead."

)

bucket = bucket_or_name

if bucket.client is None:

bucket._client = self

else:

bucket = Bucket(self, name=bucket_or_name, generation=generation)

return bucket

def get_bucket(

self,

bucket_or_name,

timeout=_DEFAULT_TIMEOUT,

if_metageneration_match=None,

if_metageneration_not_match=None,

retry=DEFAULT_RETRY,

*,

generation=None,

soft_deleted=None,

):

"""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).

Args:

bucket_or_name (Union[ \

:class:`~google.cloud.storage.bucket.Bucket`, \

str, \

]):

The bucket resource to pass or name to create.

timeout (Optional[Union[float, Tuple[float, float]]]):

The amount of time, in seconds, to wait for the server response.

Can also be passed as a tuple (connect_timeout, read_timeout).

See :meth:`requests.Session.request` documentation for details.

if_metageneration_match (Optional[int]):

Make the operation conditional on whether the

bucket's current metageneration matches the given value.

if_metageneration_not_match (Optional[int]):

Make the operation conditional on whether the bucket's

current metageneration does not match the given value.

retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):

How to retry the RPC. A None value will disable retries.

A google.api_core.retry.Retry value will enable retries, and the object will

define retriable response codes and errors and configure backoff and timeout options.

A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and

activates it only if certain conditions are met. This class exists to provide safe defaults

for RPC calls that are not technically safe to retry normally (due to potential data

duplication or other side-effects) but become safe to retry if a condition such as

if_metageneration_match is set.

See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for

information on retry types and how to configure them.

generation (Optional[int]):

The generation of the bucket. The generation can be used to

specify a specific soft-deleted version of the bucket, in

conjunction with the ``soft_deleted`` argument below. If

``soft_deleted`` is not True, the generation is unused.

soft_deleted (Optional[bool]):

If True, looks for a soft-deleted bucket. Will only return

the bucket metadata if the bucket exists and is in a

soft-deleted state. The bucket ``generation`` is required if

``soft_deleted`` is set to True.

See: https://cloud.google.com/storage/docs/soft-delete

Returns:

google.cloud.storage.bucket.Bucket

The bucket matching the name provided.

Raises:

google.cloud.exceptions.NotFound

If the bucket is not found.

"""

with create_trace_span(name="Storage.Client.getBucket"):

bucket = self._bucket_arg_to_bucket(bucket_or_name, generation=generation)

bucket.reload(

client=self,

timeout=timeout,

if_metageneration_match=if_metageneration_match,

if_metageneration_not_match=if_metageneration_not_match,

retry=retry,

soft_deleted=soft_deleted,

)

return bucket

def lookup_bucket(

self,

bucket_name,

timeout=_DEFAULT_TIMEOUT,

if_metageneration_match=None,

if_metageneration_not_match=None,

retry=DEFAULT_RETRY,

):

"""Get a bucket by name, returning None if not found.

You can use this if you would rather check for a None value

than catching a NotFound exception.

:type bucket_name: str

:param bucket_name: The name of the bucket to get.

:type timeout: float or tuple

:param timeout:

(Optional) The amount of time, in seconds, to wait

for the server response. See: :ref:`configuring_timeouts`

:type if_metageneration_match: long

:param if_metageneration_match: (Optional) Make the operation conditional on whether the

blob's current metageneration matches the given value.

:type if_metageneration_not_match: long

:param if_metageneration_not_match: (Optional) Make the operation conditional on whether the

blob's current metageneration does not match the given value.

:type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy

:param retry:

(Optional) How to retry the RPC. See: :ref:`configuring_retries`

:rtype: :class:`google.cloud.storage.bucket.Bucket` or ``NoneType``

:returns: The bucket matching the name provided or None if not found.