# 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 BigQuery API."""

from __future__ import absolute_import

from __future__ import annotations

from __future__ import division

from collections import abc as collections_abc

import copy

import datetime

import functools

import gzip

import io

import itertools

import json

import math

import os

import tempfile

import typing

from typing import (

Any,

Callable,

Dict,

IO,

Iterable,

Mapping,

List,

Optional,

Sequence,

Tuple,

Union,

)

import uuid

import warnings

import requests

from google import resumable_media # type: ignore

from google.resumable_media.requests import MultipartUpload # type: ignore

from google.resumable_media.requests import ResumableUpload

import google.api_core.client_options

import google.api_core.exceptions as core_exceptions

from google.api_core.iam import Policy

from google.api_core import page_iterator

from google.api_core import retry as retries

import google.cloud._helpers # type: ignore

from google.cloud import exceptions # pytype: disable=import-error

from google.cloud.client import ClientWithProject # type: ignore # pytype: disable=import-error

try:

from google.cloud.bigquery_storage_v1.services.big_query_read.client import (

DEFAULT_CLIENT_INFO as DEFAULT_BQSTORAGE_CLIENT_INFO,

)

except ImportError:

DEFAULT_BQSTORAGE_CLIENT_INFO = None # type: ignore

from google.auth.credentials import Credentials

from google.cloud.bigquery._http import Connection

from google.cloud.bigquery import _job_helpers

from google.cloud.bigquery import _pandas_helpers

from google.cloud.bigquery import _versions_helpers

from google.cloud.bigquery import enums

from google.cloud.bigquery import exceptions as bq_exceptions

from google.cloud.bigquery import job

from google.cloud.bigquery._helpers import _get_sub_prop

from google.cloud.bigquery._helpers import _record_field_to_json

from google.cloud.bigquery._helpers import _str_or_none

from google.cloud.bigquery._helpers import _verify_job_config_type

from google.cloud.bigquery._helpers import _get_bigquery_host

from google.cloud.bigquery._helpers import _DEFAULT_HOST

from google.cloud.bigquery._helpers import _DEFAULT_HOST_TEMPLATE

from google.cloud.bigquery._helpers import _DEFAULT_UNIVERSE

from google.cloud.bigquery._helpers import _validate_universe

from google.cloud.bigquery._helpers import _get_client_universe

from google.cloud.bigquery._helpers import TimeoutType

from google.cloud.bigquery._job_helpers import make_job_id as _make_job_id

from google.cloud.bigquery.dataset import Dataset

from google.cloud.bigquery.dataset import DatasetListItem

from google.cloud.bigquery.dataset import DatasetReference

from google.cloud.bigquery.enums import AutoRowIDs, DatasetView, UpdateMode

from google.cloud.bigquery.format_options import ParquetOptions

from google.cloud.bigquery.job import (

CopyJob,

CopyJobConfig,

ExtractJob,

ExtractJobConfig,

LoadJob,

LoadJobConfig,

QueryJob,

QueryJobConfig,

)

from google.cloud.bigquery.model import Model

from google.cloud.bigquery.model import ModelReference

from google.cloud.bigquery.model import _model_arg_to_model_ref

from google.cloud.bigquery.opentelemetry_tracing import create_span

from google.cloud.bigquery.query import _QueryResults

from google.cloud.bigquery.retry import (

DEFAULT_JOB_RETRY,

DEFAULT_RETRY,

DEFAULT_TIMEOUT,

DEFAULT_GET_JOB_TIMEOUT,

POLLING_DEFAULT_VALUE,

)

from google.cloud.bigquery.routine import Routine

from google.cloud.bigquery.routine import RoutineReference

from google.cloud.bigquery.schema import SchemaField

from google.cloud.bigquery.table import _table_arg_to_table

from google.cloud.bigquery.table import _table_arg_to_table_ref

from google.cloud.bigquery.table import Table

from google.cloud.bigquery.table import TableListItem

from google.cloud.bigquery.table import TableReference

from google.cloud.bigquery.table import RowIterator

pyarrow = _versions_helpers.PYARROW_VERSIONS.try_import()

pandas = (

_versions_helpers.PANDAS_VERSIONS.try_import()

) # mypy check fails because pandas import is outside module, there are type: ignore comments related to this

ResumableTimeoutType = Union[

None, float, Tuple[float, float]

] # for resumable media methods

if typing.TYPE_CHECKING: # pragma: NO COVER

# os.PathLike is only subscriptable in Python 3.9+, thus shielding with a condition.

PathType = Union[str, bytes, os.PathLike[str], os.PathLike[bytes], io.IOBase]

_DEFAULT_CHUNKSIZE = 100 * 1024 * 1024 # 100 MB

_MAX_MULTIPART_SIZE = 5 * 1024 * 1024

_DEFAULT_NUM_RETRIES = 6

_BASE_UPLOAD_TEMPLATE = "{host}/upload/bigquery/v2/projects/{project}/jobs?uploadType="

_MULTIPART_URL_TEMPLATE = _BASE_UPLOAD_TEMPLATE + "multipart"

_RESUMABLE_URL_TEMPLATE = _BASE_UPLOAD_TEMPLATE + "resumable"

_GENERIC_CONTENT_TYPE = "*/*"

_READ_LESS_THAN_SIZE = (

"Size {:d} was specified but the file-like object only had " "{:d} bytes remaining."

)

_NEED_TABLE_ARGUMENT = (

"The table argument should be a table ID string, Table, or TableReference"

)

_LIST_ROWS_FROM_QUERY_RESULTS_FIELDS = "jobReference,totalRows,pageToken,rows"

# In microbenchmarks, it's been shown that even in ideal conditions (query

# finished, local data), requests to getQueryResults can take 10+ seconds.

# In less-than-ideal situations, the response can take even longer, as it must

# be able to download a full 100+ MB row in that time. Don't let the

# connection timeout before data can be downloaded.

# https://github.com/googleapis/python-bigquery/issues/438

_MIN_GET_QUERY_RESULTS_TIMEOUT = 120

TIMEOUT_HEADER = "X-Server-Timeout"

class Project(object):

"""Wrapper for resource describing a BigQuery project.

Args:

project_id (str): Opaque ID of the project

numeric_id (int): Numeric ID of the project

friendly_name (str): Display name of the project

"""

def __init__(self, project_id, numeric_id, friendly_name):

self.project_id = project_id

self.numeric_id = numeric_id

self.friendly_name = friendly_name

@classmethod

def from_api_repr(cls, resource):

"""Factory: construct an instance from a resource dict."""

return cls(resource["id"], resource["numericId"], resource["friendlyName"])

class Client(ClientWithProject):

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

Args:

project (Optional[str]):

Project ID for the project which the client acts on behalf of.

Will be passed when creating a dataset / job. If not passed,

falls back to the default inferred from the environment.

credentials (Optional[google.auth.credentials.Credentials]):

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.

_http (Optional[requests.Session]):

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.

location (Optional[str]):

Default location for jobs / datasets / tables.

default_query_job_config (Optional[google.cloud.bigquery.job.QueryJobConfig]):

Default ``QueryJobConfig``.

Will be merged into job configs passed into the ``query`` method.

default_load_job_config (Optional[google.cloud.bigquery.job.LoadJobConfig]):

Default ``LoadJobConfig``.

Will be merged into job configs passed into the ``load_table_*`` methods.

client_info (Optional[google.api_core.client_info.ClientInfo]):

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.

client_options (Optional[Union[google.api_core.client_options.ClientOptions, Dict]]):

Client options used to set user options on the client. API Endpoint

should be set through client_options.

default_job_creation_mode (Optional[str]):

Sets the default job creation mode used by query methods such as

query_and_wait(). For lightweight queries, JOB_CREATION_OPTIONAL is

generally recommended.

Raises:

google.auth.exceptions.DefaultCredentialsError:

Raised if ``credentials`` is not specified and the library fails

to acquire default credentials.

"""

SCOPE = ("https://www.googleapis.com/auth/cloud-platform",) # type: ignore

"""The scopes required for authenticating as a BigQuery consumer."""

def __init__(

self,

project: Optional[str] = None,

credentials: Optional[Credentials] = None,

_http: Optional[requests.Session] = None,

location: Optional[str] = None,

default_query_job_config: Optional[QueryJobConfig] = None,

default_load_job_config: Optional[LoadJobConfig] = None,

client_info: Optional[google.api_core.client_info.ClientInfo] = None,

client_options: Optional[

Union[google.api_core.client_options.ClientOptions, Dict[str, Any]]

] = None,

default_job_creation_mode: Optional[str] = None,

) -> None:

if client_options is None:

client_options = {}

if isinstance(client_options, dict):

client_options = google.api_core.client_options.from_dict(client_options)

# assert isinstance(client_options, google.api_core.client_options.ClientOptions)

super(Client, self).__init__(

project=project,

credentials=credentials,

client_options=client_options,

_http=_http,

)

kw_args: Dict[str, Any] = {"client_info": client_info}

bq_host = _get_bigquery_host()

kw_args["api_endpoint"] = bq_host if bq_host != _DEFAULT_HOST else None

client_universe = None

if client_options.api_endpoint:

api_endpoint = client_options.api_endpoint

kw_args["api_endpoint"] = api_endpoint

else:

client_universe = _get_client_universe(client_options)

if client_universe != _DEFAULT_UNIVERSE:

kw_args["api_endpoint"] = _DEFAULT_HOST_TEMPLATE.replace(

"{UNIVERSE_DOMAIN}", client_universe

)

# Ensure credentials and universe are not in conflict.

if hasattr(self, "_credentials") and client_universe is not None:

_validate_universe(client_universe, self._credentials)

self._connection = Connection(self, **kw_args)

self._location = location

self._default_load_job_config = copy.deepcopy(default_load_job_config)

self.default_job_creation_mode = default_job_creation_mode

# Use property setter so validation can run.

self.default_query_job_config = default_query_job_config

@property

def location(self):

"""Default location for jobs / datasets / tables."""

return self._location

@property

def default_job_creation_mode(self):

"""Default job creation mode used for query execution."""

return self._default_job_creation_mode

@default_job_creation_mode.setter

def default_job_creation_mode(self, value: Optional[str]):

self._default_job_creation_mode = value

@property

def default_query_job_config(self) -> Optional[QueryJobConfig]:

"""Default ``QueryJobConfig`` or ``None``.

Will be merged into job configs passed into the ``query`` or

``query_and_wait`` methods.

"""

return self._default_query_job_config

@default_query_job_config.setter

def default_query_job_config(self, value: Optional[QueryJobConfig]):

if value is not None:

_verify_job_config_type(

value, QueryJobConfig, param_name="default_query_job_config"

)

self._default_query_job_config = copy.deepcopy(value)

@property

def default_load_job_config(self):

"""Default ``LoadJobConfig``.

Will be merged into job configs passed into the ``load_table_*`` methods.

"""

return self._default_load_job_config

@default_load_job_config.setter

def default_load_job_config(self, value: LoadJobConfig):

self._default_load_job_config = copy.deepcopy(value)

def close(self):

"""Close the underlying transport objects, releasing system resources.

.. note::

The client instance can be used for making additional requests even

after closing, in which case the underlying connections are

automatically re-created.

"""

self._http._auth_request.session.close()

self._http.close()

def get_service_account_email(

self,

project: Optional[str] = None,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

) -> str:

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

Example:

.. code-block:: python

from google.cloud import bigquery

client = bigquery.Client()

client.get_service_account_email()

# returns an email similar to: my_service_account@my-project.iam.gserviceaccount.com

Note:

This is the service account that BigQuery uses to manage tables

encrypted by a key in KMS.

Args:

project (Optional[str]):

Project ID to use for retreiving service account email.

Defaults to the client's project.

retry (Optional[google.api_core.retry.Retry]): How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

Returns:

str:

service account email address

"""

if project is None:

project = self.project

path = "/projects/%s/serviceAccount" % (project,)

span_attributes = {"path": path}

api_response = self._call_api(

retry,

span_name="BigQuery.getServiceAccountEmail",

span_attributes=span_attributes,

method="GET",

path=path,

timeout=timeout,

)

return api_response["email"]

def list_projects(

self,

max_results: Optional[int] = None,

page_token: Optional[str] = None,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

page_size: Optional[int] = None,

) -> page_iterator.Iterator:

"""List projects for the project associated with this client.

See

https://cloud.google.com/bigquery/docs/reference/rest/v2/projects/list

Args:

max_results (Optional[int]):

Maximum number of projects to return.

Defaults to a value set by the API.

page_token (Optional[str]):

Token representing a cursor into the projects. If not passed,

the API will return the first page of projects. The token marks

the beginning of the iterator to be returned and the value of

the ``page_token`` can be accessed at ``next_page_token`` of the

:class:`~google.api_core.page_iterator.HTTPIterator`.

retry (Optional[google.api_core.retry.Retry]): How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

page_size (Optional[int]):

Maximum number of projects to return in each page.

Defaults to a value set by the API.

Returns:

google.api_core.page_iterator.Iterator:

Iterator of :class:`~google.cloud.bigquery.client.Project`

accessible to the current client.

"""

span_attributes = {"path": "/projects"}

def api_request(*args, **kwargs):

return self._call_api(

retry,

span_name="BigQuery.listProjects",

span_attributes=span_attributes,

*args,

timeout=timeout,

**kwargs,

)

return page_iterator.HTTPIterator(

client=self,

api_request=api_request,

path="/projects",

item_to_value=_item_to_project,

items_key="projects",

page_token=page_token,

max_results=max_results,

page_size=page_size,

)

def list_datasets(

self,

project: Optional[str] = None,

include_all: bool = False,

filter: Optional[str] = None,

max_results: Optional[int] = None,

page_token: Optional[str] = None,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

page_size: Optional[int] = None,

) -> page_iterator.Iterator:

"""List datasets for the project associated with this client.

See

https://cloud.google.com/bigquery/docs/reference/rest/v2/datasets/list

Args:

project (Optional[str]):

Project ID to use for retreiving datasets. Defaults to the

client's project.

include_all (Optional[bool]):

True if results include hidden datasets. Defaults to False.

filter (Optional[str]):

An expression for filtering the results by label.

For syntax, see

https://cloud.google.com/bigquery/docs/reference/rest/v2/datasets/list#body.QUERY_PARAMETERS.filter

max_results (Optional[int]):

Maximum number of datasets to return.

page_token (Optional[str]):

Token representing a cursor into the datasets. If not passed,

the API will return the first page of datasets. The token marks

the beginning of the iterator to be returned and the value of

the ``page_token`` can be accessed at ``next_page_token`` of the

:class:`~google.api_core.page_iterator.HTTPIterator`.

retry (Optional[google.api_core.retry.Retry]):

How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

page_size (Optional[int]):

Maximum number of datasets to return per page.

Returns:

google.api_core.page_iterator.Iterator:

Iterator of :class:`~google.cloud.bigquery.dataset.DatasetListItem`.

associated with the project.

"""

extra_params: Dict[str, Any] = {}

if project is None:

project = self.project

if include_all:

extra_params["all"] = True

if filter:

# TODO: consider supporting a dict of label -> value for filter,

# and converting it into a string here.

extra_params["filter"] = filter

path = "/projects/%s/datasets" % (project,)

span_attributes = {"path": path}

def api_request(*args, **kwargs):

return self._call_api(

retry,

span_name="BigQuery.listDatasets",

span_attributes=span_attributes,

*args,

timeout=timeout,

**kwargs,

)

return page_iterator.HTTPIterator(

client=self,

api_request=api_request,

path=path,

item_to_value=_item_to_dataset,

items_key="datasets",

page_token=page_token,

max_results=max_results,

extra_params=extra_params,

page_size=page_size,

)

def dataset(

self, dataset_id: str, project: Optional[str] = None

) -> DatasetReference:

"""Deprecated: Construct a reference to a dataset.

.. deprecated:: 1.24.0

Construct a

:class:`~google.cloud.bigquery.dataset.DatasetReference` using its

constructor or use a string where previously a reference object

was used.

As of ``google-cloud-bigquery`` version 1.7.0, all client methods

that take a

:class:`~google.cloud.bigquery.dataset.DatasetReference` or

:class:`~google.cloud.bigquery.table.TableReference` also take a

string in standard SQL format, e.g. ``project.dataset_id`` or

``project.dataset_id.table_id``.

Args:

dataset_id (str): ID of the dataset.

project (Optional[str]):

Project ID for the dataset (defaults to the project of the client).

Returns:

google.cloud.bigquery.dataset.DatasetReference:

a new ``DatasetReference`` instance.

"""

if project is None:

project = self.project

warnings.warn(

"Client.dataset is deprecated and will be removed in a future version. "

"Use a string like 'my_project.my_dataset' or a "

"cloud.google.bigquery.DatasetReference object, instead.",

PendingDeprecationWarning,

stacklevel=2,

)

return DatasetReference(project, dataset_id)

def _ensure_bqstorage_client(

self,

bqstorage_client: Optional[

"google.cloud.bigquery_storage.BigQueryReadClient"

] = None,

client_options: Optional[google.api_core.client_options.ClientOptions] = None,

client_info: Optional[

"google.api_core.gapic_v1.client_info.ClientInfo"

] = DEFAULT_BQSTORAGE_CLIENT_INFO,

) -> Optional["google.cloud.bigquery_storage.BigQueryReadClient"]:

"""Create a BigQuery Storage API client using this client's credentials.

Args:

bqstorage_client:

An existing BigQuery Storage client instance. If ``None``, a new

instance is created and returned.

client_options:

Custom options used with a new BigQuery Storage client instance

if one is created.

client_info:

The client info used with a new BigQuery Storage client

instance if one is created.

Returns:

A BigQuery Storage API client.

"""

try:

bigquery_storage = _versions_helpers.BQ_STORAGE_VERSIONS.try_import(

raise_if_error=True

)

except bq_exceptions.BigQueryStorageNotFoundError:

warnings.warn(

"Cannot create BigQuery Storage client, the dependency "

"google-cloud-bigquery-storage is not installed."

)

return None

except bq_exceptions.LegacyBigQueryStorageError as exc:

warnings.warn(

"Dependency google-cloud-bigquery-storage is outdated: " + str(exc)

)

return None

if bqstorage_client is None: # pragma: NO COVER

bqstorage_client = bigquery_storage.BigQueryReadClient(

credentials=self._credentials,

client_options=client_options,

client_info=client_info, # type: ignore # (None is also accepted)

)

return bqstorage_client

def _dataset_from_arg(self, dataset) -> Union[Dataset, DatasetReference]:

if isinstance(dataset, str):

dataset = DatasetReference.from_string(

dataset, default_project=self.project

)

if not isinstance(dataset, (Dataset, DatasetReference)):

if isinstance(dataset, DatasetListItem):

dataset = dataset.reference

else:

raise TypeError(

"dataset must be a Dataset, DatasetReference, DatasetListItem,"

" or string"

)

return dataset

def create_dataset(

self,

dataset: Union[str, Dataset, DatasetReference, DatasetListItem],

exists_ok: bool = False,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

) -> Dataset:

"""API call: create the dataset via a POST request.

See

https://cloud.google.com/bigquery/docs/reference/rest/v2/datasets/insert

Example:

.. code-block:: python

from google.cloud import bigquery

client = bigquery.Client()

dataset = bigquery.Dataset('my_project.my_dataset')

dataset = client.create_dataset(dataset)

Args:

dataset (Union[ \

google.cloud.bigquery.dataset.Dataset, \

google.cloud.bigquery.dataset.DatasetReference, \

google.cloud.bigquery.dataset.DatasetListItem, \

str, \

]):

A :class:`~google.cloud.bigquery.dataset.Dataset` to create.

If ``dataset`` is a reference, an empty dataset is created

with the specified ID and client's default location.

exists_ok (Optional[bool]):

Defaults to ``False``. If ``True``, ignore "already exists"

errors when creating the dataset.

retry (Optional[google.api_core.retry.Retry]):

How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

Returns:

google.cloud.bigquery.dataset.Dataset:

A new ``Dataset`` returned from the API.

Raises:

google.cloud.exceptions.Conflict:

If the dataset already exists.

"""

dataset = self._dataset_from_arg(dataset)

if isinstance(dataset, DatasetReference):

dataset = Dataset(dataset)

path = "/projects/%s/datasets" % (dataset.project,)

data = dataset.to_api_repr()

if data.get("location") is None and self.location is not None:

data["location"] = self.location

try:

span_attributes = {"path": path}

api_response = self._call_api(

retry,

span_name="BigQuery.createDataset",

span_attributes=span_attributes,

method="POST",

path=path,

data=data,

timeout=timeout,

)

return Dataset.from_api_repr(api_response)

except core_exceptions.Conflict:

if not exists_ok:

raise

return self.get_dataset(dataset.reference, retry=retry)

def create_routine(

self,

routine: Routine,

exists_ok: bool = False,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

) -> Routine:

"""[Beta] Create a routine via a POST request.

See

https://cloud.google.com/bigquery/docs/reference/rest/v2/routines/insert

Args:

routine (google.cloud.bigquery.routine.Routine):

A :class:`~google.cloud.bigquery.routine.Routine` to create.

The dataset that the routine belongs to must already exist.

exists_ok (Optional[bool]):

Defaults to ``False``. If ``True``, ignore "already exists"

errors when creating the routine.

retry (Optional[google.api_core.retry.Retry]):

How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

Returns:

google.cloud.bigquery.routine.Routine:

A new ``Routine`` returned from the service.

Raises:

google.cloud.exceptions.Conflict:

If the routine already exists.

"""

reference = routine.reference

path = "/projects/{}/datasets/{}/routines".format(

reference.project, reference.dataset_id

)

resource = routine.to_api_repr()

try:

span_attributes = {"path": path}

api_response = self._call_api(

retry,

span_name="BigQuery.createRoutine",

span_attributes=span_attributes,

method="POST",

path=path,

data=resource,

timeout=timeout,

)

return Routine.from_api_repr(api_response)

except core_exceptions.Conflict:

if not exists_ok:

raise

return self.get_routine(routine.reference, retry=retry)

def create_table(

self,

table: Union[str, Table, TableReference, TableListItem],

exists_ok: bool = False,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

) -> Table:

"""API call: create a table via a PUT request

See

https://cloud.google.com/bigquery/docs/reference/rest/v2/tables/insert

Args:

table (Union[ \

google.cloud.bigquery.table.Table, \

google.cloud.bigquery.table.TableReference, \

google.cloud.bigquery.table.TableListItem, \

str, \

]):

A :class:`~google.cloud.bigquery.table.Table` to create.

If ``table`` is a reference, an empty table is created

with the specified ID. The dataset that the table belongs to

must already exist.

exists_ok (Optional[bool]):

Defaults to ``False``. If ``True``, ignore "already exists"

errors when creating the table.

retry (Optional[google.api_core.retry.Retry]):

How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

Returns:

google.cloud.bigquery.table.Table:

A new ``Table`` returned from the service.

Raises:

google.cloud.exceptions.Conflict:

If the table already exists.

"""

table = _table_arg_to_table(table, default_project=self.project)

dataset_id = table.dataset_id

path = "/projects/%s/datasets/%s/tables" % (table.project, dataset_id)

data = table.to_api_repr()

try:

span_attributes = {"path": path, "dataset_id": dataset_id}

api_response = self._call_api(

retry,

span_name="BigQuery.createTable",

span_attributes=span_attributes,

method="POST",

path=path,

data=data,

timeout=timeout,

)

return Table.from_api_repr(api_response)

except core_exceptions.Conflict:

if not exists_ok:

raise

return self.get_table(table.reference, retry=retry)

def _call_api(

self,

retry,

span_name=None,

span_attributes=None,

job_ref=None,

headers: Optional[Dict[str, str]] = None,

**kwargs,

):

kwargs = _add_server_timeout_header(headers, kwargs)

call = functools.partial(self._connection.api_request, **kwargs)

if retry:

call = retry(call)

if span_name is not None:

with create_span(

name=span_name, attributes=span_attributes, client=self, job_ref=job_ref

):

return call()

return call()

def get_dataset(

self,

dataset_ref: Union[DatasetReference, str],

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

dataset_view: Optional[DatasetView] = None,

) -> Dataset:

"""Fetch the dataset referenced by ``dataset_ref``

Args:

dataset_ref (Union[ \

google.cloud.bigquery.dataset.DatasetReference, \

str, \

]):

A reference to the dataset to fetch from the BigQuery API.

If a string is passed in, this method attempts to create a

dataset reference from a string using

:func:`~google.cloud.bigquery.dataset.DatasetReference.from_string`.

retry (Optional[google.api_core.retry.Retry]):

How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

dataset_view (Optional[google.cloud.bigquery.enums.DatasetView]):

Specifies the view that determines which dataset information is

returned. By default, dataset metadata (e.g. friendlyName, description,

labels, etc) and ACL information are returned. This argument can

take on the following possible enum values.

* :attr:`~google.cloud.bigquery.enums.DatasetView.ACL`:

Includes dataset metadata and the ACL.

* :attr:`~google.cloud.bigquery.enums.DatasetView.FULL`:

Includes all dataset metadata, including the ACL and table metadata.

This view is not supported by the `datasets.list` API method.

* :attr:`~google.cloud.bigquery.enums.DatasetView.METADATA`:

Includes basic dataset metadata, but not the ACL.

* :attr:`~google.cloud.bigquery.enums.DatasetView.DATASET_VIEW_UNSPECIFIED`:

The server will decide which view to use. Currently defaults to FULL.

Returns:

google.cloud.bigquery.dataset.Dataset:

A ``Dataset`` instance.

"""

if isinstance(dataset_ref, str):

dataset_ref = DatasetReference.from_string(

dataset_ref, default_project=self.project

)

path = dataset_ref.path

if dataset_view:

query_params = {"datasetView": dataset_view.value}

else:

query_params = {}

span_attributes = {"path": path}

api_response = self._call_api(

retry,

span_name="BigQuery.getDataset",

span_attributes=span_attributes,

method="GET",

path=path,

timeout=timeout,

query_params=query_params,

)

return Dataset.from_api_repr(api_response)

def get_iam_policy(

self,

table: Union[Table, TableReference, TableListItem, str],

requested_policy_version: int = 1,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

) -> Policy:

"""Return the access control policy for a table resource.

Args:

table (Union[ \

google.cloud.bigquery.table.Table, \

google.cloud.bigquery.table.TableReference, \

google.cloud.bigquery.table.TableListItem, \

str, \

]):

The table to get the access control policy for.

If a string is passed in, this method attempts to create a

table reference from a string using

:func:`~google.cloud.bigquery.table.TableReference.from_string`.

requested_policy_version (int):

Optional. The maximum policy version that will be used to format the policy.

Only version ``1`` is currently supported.

See: https://cloud.google.com/bigquery/docs/reference/rest/v2/GetPolicyOptions

retry (Optional[google.api_core.retry.Retry]):

How to retry the RPC.

timeout (Optional[float]):

The number of seconds to wait for the underlying HTTP transport

before using ``retry``.

Returns:

google.api_core.iam.Policy:

The access control policy.

"""

table = _table_arg_to_table_ref(table, default_project=self.project)

if requested_policy_version != 1:

raise ValueError("only IAM policy version 1 is supported")

body = {"options": {"requestedPolicyVersion": 1}}

path = "{}:getIamPolicy".format(table.path)

span_attributes = {"path": path}

response = self._call_api(

retry,

span_name="BigQuery.getIamPolicy",

span_attributes=span_attributes,

method="POST",

path=path,

data=body,

timeout=timeout,

)

return Policy.from_api_repr(response)

def set_iam_policy(

self,

table: Union[Table, TableReference, TableListItem, str],

policy: Policy,

updateMask: Optional[str] = None,

retry: retries.Retry = DEFAULT_RETRY,

timeout: TimeoutType = DEFAULT_TIMEOUT,

*,

fields: Sequence[str] = (),

) -> Policy:

"""Return the access control policy for a table resource.

Args:

table (Union[ \