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

"""Testable usage examples for Google BigQuery API wrapper

Each example function takes a ``client`` argument (which must be an instance

of :class:`google.cloud.bigquery.client.Client`) and uses it to perform a task

with the API.

To facilitate running the examples as system tests, each example is also passed

a ``to_delete`` list; the function adds to the list any objects created which

need to be deleted during teardown.

"""

import os

import time

import mock

import pytest

import six

try:

import fastparquet

except (ImportError, AttributeError):

fastparquet = None

try:

import pandas

except (ImportError, AttributeError):

pandas = None

try:

import pyarrow

except (ImportError, AttributeError):

pyarrow = None

from google.api_core import datetime_helpers

from google.api_core.exceptions import InternalServerError

from google.api_core.exceptions import ServiceUnavailable

from google.api_core.exceptions import TooManyRequests

from google.cloud import bigquery

from google.cloud import storage

from test_utils.retry import RetryErrors

ORIGINAL_FRIENDLY_NAME = "Original friendly name"

ORIGINAL_DESCRIPTION = "Original description"

LOCALLY_CHANGED_FRIENDLY_NAME = "Locally-changed friendly name"

LOCALLY_CHANGED_DESCRIPTION = "Locally-changed description"

UPDATED_FRIENDLY_NAME = "Updated friendly name"

UPDATED_DESCRIPTION = "Updated description"

SCHEMA = [

bigquery.SchemaField("full_name", "STRING", mode="REQUIRED"),

bigquery.SchemaField("age", "INTEGER", mode="REQUIRED"),

]

ROWS = [

("Phred Phlyntstone", 32),

("Bharney Rhubble", 33),

("Wylma Phlyntstone", 29),

("Bhettye Rhubble", 27),

]

QUERY = (

"SELECT name FROM `bigquery-public-data.usa_names.usa_1910_2013` "

'WHERE state = "TX"'

)

retry_429 = RetryErrors(TooManyRequests)

retry_storage_errors = RetryErrors(

(TooManyRequests, InternalServerError, ServiceUnavailable)

)

@pytest.fixture(scope="module")

def client():

return bigquery.Client()

@pytest.fixture

def to_delete(client):

doomed = []

yield doomed

for item in doomed:

if isinstance(item, (bigquery.Dataset, bigquery.DatasetReference)):

retry_429(client.delete_dataset)(item, delete_contents=True)

elif isinstance(item, storage.Bucket):

retry_storage_errors(item.delete)()

else:

retry_429(item.delete)()

def _millis():

return int(time.time() * 1000)

class _CloseOnDelete(object):

def __init__(self, wrapped):

self._wrapped = wrapped

def delete(self):

self._wrapped.close()

def test_create_client_default_credentials():

"""Create a BigQuery client with Application Default Credentials"""

# [START bigquery_client_default_credentials]

from google.cloud import bigquery

# If you don't specify credentials when constructing the client, the

# client library will look for credentials in the environment.

client = bigquery.Client()

# [END bigquery_client_default_credentials]

assert client is not None

def test_create_client_json_credentials():

"""Create a BigQuery client with Application Default Credentials"""

with open(os.environ["GOOGLE_APPLICATION_CREDENTIALS"]) as creds_file:

creds_file_data = creds_file.read()

open_mock = mock.mock_open(read_data=creds_file_data)

with mock.patch("io.open", open_mock):

# [START bigquery_client_json_credentials]

from google.cloud import bigquery

# Explicitly use service account credentials by specifying the private

# key file. All clients in google-cloud-python have this helper.

client = bigquery.Client.from_service_account_json(

"path/to/service_account.json"

)

# [END bigquery_client_json_credentials]

assert client is not None

def test_list_datasets_by_label(client, to_delete):

dataset_id = "list_datasets_by_label_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

dataset.labels = {"color": "green"}

dataset = client.create_dataset(dataset) # API request

to_delete.append(dataset)

# [START bigquery_list_datasets_by_label]

# from google.cloud import bigquery

# client = bigquery.Client()

# The following label filter example will find datasets with an

# arbitrary 'color' label set to 'green'

label_filter = "labels.color:green"

datasets = list(client.list_datasets(filter=label_filter))

if datasets:

print("Datasets filtered by {}:".format(label_filter))

for dataset in datasets: # API request(s)

print("\t{}".format(dataset.dataset_id))

else:

print("No datasets found with this filter.")

# [END bigquery_list_datasets_by_label]

found = set([dataset.dataset_id for dataset in datasets])

assert dataset_id in found

# [START bigquery_dataset_exists]

def dataset_exists(client, dataset_reference):

"""Return if a dataset exists.

Args:

client (google.cloud.bigquery.client.Client):

A client to connect to the BigQuery API.

dataset_reference (google.cloud.bigquery.dataset.DatasetReference):

A reference to the dataset to look for.

Returns:

bool: ``True`` if the dataset exists, ``False`` otherwise.

"""

from google.cloud.exceptions import NotFound

try:

client.get_dataset(dataset_reference)

return True

except NotFound:

return False

# [END bigquery_dataset_exists]

def test_dataset_exists(client, to_delete):

"""Determine if a dataset exists."""

DATASET_ID = "get_table_dataset_{}".format(_millis())

dataset_ref = client.dataset(DATASET_ID)

dataset = bigquery.Dataset(dataset_ref)

dataset = client.create_dataset(dataset)

to_delete.append(dataset)

assert dataset_exists(client, dataset_ref)

assert not dataset_exists(client, client.dataset("i_dont_exist"))

@pytest.mark.skip(

reason=(

"update_dataset() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5588"

)

)

def test_manage_dataset_labels(client, to_delete):

dataset_id = "label_dataset_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

dataset = client.create_dataset(dataset)

to_delete.append(dataset)

# [START bigquery_label_dataset]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_ref = client.dataset('my_dataset')

# dataset = client.get_dataset(dataset_ref) # API request

assert dataset.labels == {}

labels = {"color": "green"}

dataset.labels = labels

dataset = client.update_dataset(dataset, ["labels"]) # API request

assert dataset.labels == labels

# [END bigquery_label_dataset]

# [START bigquery_get_dataset_labels]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset'

dataset_ref = client.dataset(dataset_id)

dataset = client.get_dataset(dataset_ref) # API request

# View dataset labels

print("Dataset ID: {}".format(dataset_id))

print("Labels:")

if dataset.labels:

for label, value in dataset.labels.items():

print("\t{}: {}".format(label, value))

else:

print("\tDataset has no labels defined.")

# [END bigquery_get_dataset_labels]

assert dataset.labels == labels

# [START bigquery_delete_label_dataset]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_ref = client.dataset('my_dataset')

# dataset = client.get_dataset(dataset_ref) # API request

# This example dataset starts with one label

assert dataset.labels == {"color": "green"}

# To delete a label from a dataset, set its value to None

dataset.labels["color"] = None

dataset = client.update_dataset(dataset, ["labels"]) # API request

assert dataset.labels == {}

# [END bigquery_delete_label_dataset]

def test_create_table_nested_repeated_schema(client, to_delete):

dataset_id = "create_table_nested_repeated_{}".format(_millis())

dataset_ref = client.dataset(dataset_id)

dataset = bigquery.Dataset(dataset_ref)

client.create_dataset(dataset)

to_delete.append(dataset)

# [START bigquery_nested_repeated_schema]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_ref = client.dataset('my_dataset')

schema = [

bigquery.SchemaField("id", "STRING", mode="NULLABLE"),

bigquery.SchemaField("first_name", "STRING", mode="NULLABLE"),

bigquery.SchemaField("last_name", "STRING", mode="NULLABLE"),

bigquery.SchemaField("dob", "DATE", mode="NULLABLE"),

bigquery.SchemaField(

"addresses",

"RECORD",

mode="REPEATED",

fields=[

bigquery.SchemaField("status", "STRING", mode="NULLABLE"),

bigquery.SchemaField("address", "STRING", mode="NULLABLE"),

bigquery.SchemaField("city", "STRING", mode="NULLABLE"),

bigquery.SchemaField("state", "STRING", mode="NULLABLE"),

bigquery.SchemaField("zip", "STRING", mode="NULLABLE"),

bigquery.SchemaField("numberOfYears", "STRING", mode="NULLABLE"),

],

),

]

table_ref = dataset_ref.table("my_table")

table = bigquery.Table(table_ref, schema=schema)

table = client.create_table(table) # API request

print("Created table {}".format(table.full_table_id))

# [END bigquery_nested_repeated_schema]

def test_create_table_cmek(client, to_delete):

dataset_id = "create_table_cmek_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

client.create_dataset(dataset)

to_delete.append(dataset)

# [START bigquery_create_table_cmek]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset'

table_ref = client.dataset(dataset_id).table("my_table")

table = bigquery.Table(table_ref)

# Set the encryption key to use for the table.

# TODO: Replace this key with a key you have created in Cloud KMS.

kms_key_name = "projects/{}/locations/{}/keyRings/{}/cryptoKeys/{}".format(

"cloud-samples-tests", "us-central1", "test", "test"

)

table.encryption_configuration = bigquery.EncryptionConfiguration(

kms_key_name=kms_key_name

)

table = client.create_table(table) # API request

assert table.encryption_configuration.kms_key_name == kms_key_name

# [END bigquery_create_table_cmek]

def test_create_partitioned_table(client, to_delete):

dataset_id = "create_table_partitioned_{}".format(_millis())

dataset_ref = bigquery.Dataset(client.dataset(dataset_id))

dataset = client.create_dataset(dataset_ref)

to_delete.append(dataset)

# [START bigquery_create_table_partitioned]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_ref = client.dataset('my_dataset')

table_ref = dataset_ref.table("my_partitioned_table")

schema = [

bigquery.SchemaField("name", "STRING"),

bigquery.SchemaField("post_abbr", "STRING"),

bigquery.SchemaField("date", "DATE"),

]

table = bigquery.Table(table_ref, schema=schema)

table.time_partitioning = bigquery.TimePartitioning(

type_=bigquery.TimePartitioningType.DAY,

field="date", # name of column to use for partitioning

expiration_ms=7776000000,

) # 90 days

table = client.create_table(table)

print(

"Created table {}, partitioned on column {}".format(

table.table_id, table.time_partitioning.field

)

)

# [END bigquery_create_table_partitioned]

assert table.time_partitioning.type_ == "DAY"

assert table.time_partitioning.field == "date"

assert table.time_partitioning.expiration_ms == 7776000000

def test_load_and_query_partitioned_table(client, to_delete):

dataset_id = "load_partitioned_table_dataset_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

client.create_dataset(dataset)

to_delete.append(dataset)

# [START bigquery_load_table_partitioned]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset'

table_id = "us_states_by_date"

dataset_ref = client.dataset(dataset_id)

job_config = bigquery.LoadJobConfig()

job_config.schema = [

bigquery.SchemaField("name", "STRING"),

bigquery.SchemaField("post_abbr", "STRING"),

bigquery.SchemaField("date", "DATE"),

]

job_config.skip_leading_rows = 1

job_config.time_partitioning = bigquery.TimePartitioning(

type_=bigquery.TimePartitioningType.DAY,

field="date", # name of column to use for partitioning

expiration_ms=7776000000,

) # 90 days

uri = "gs://cloud-samples-data/bigquery/us-states/us-states-by-date.csv"

load_job = client.load_table_from_uri(

uri, dataset_ref.table(table_id), job_config=job_config

) # API request

assert load_job.job_type == "load"

load_job.result() # Waits for table load to complete.

table = client.get_table(dataset_ref.table(table_id))

print("Loaded {} rows to table {}".format(table.num_rows, table_id))

# [END bigquery_load_table_partitioned]

assert table.num_rows == 50

project_id = client.project

# [START bigquery_query_partitioned_table]

import datetime

# from google.cloud import bigquery

# client = bigquery.Client()

# project_id = 'my-project'

# dataset_id = 'my_dataset'

table_id = "us_states_by_date"

sql_template = """

SELECT *

FROM `{}.{}.{}`

WHERE date BETWEEN @start_date AND @end_date

"""

sql = sql_template.format(project_id, dataset_id, table_id)

job_config = bigquery.QueryJobConfig()

job_config.query_parameters = [

bigquery.ScalarQueryParameter("start_date", "DATE", datetime.date(1800, 1, 1)),

bigquery.ScalarQueryParameter("end_date", "DATE", datetime.date(1899, 12, 31)),

]

query_job = client.query(

sql,

# Location must match that of the dataset(s) referenced in the query.

location="US",

job_config=job_config,

) # API request

rows = list(query_job)

print("{} states were admitted to the US in the 1800s".format(len(rows)))

# [END bigquery_query_partitioned_table]

assert len(rows) == 29

# [START bigquery_table_exists]

def table_exists(client, table_reference):

"""Return if a table exists.

Args:

client (google.cloud.bigquery.client.Client):

A client to connect to the BigQuery API.

table_reference (google.cloud.bigquery.table.TableReference):

A reference to the table to look for.

Returns:

bool: ``True`` if the table exists, ``False`` otherwise.

"""

from google.cloud.exceptions import NotFound

try:

client.get_table(table_reference)

return True

except NotFound:

return False

# [END bigquery_table_exists]

def test_table_exists(client, to_delete):

"""Determine if a table exists."""

DATASET_ID = "get_table_dataset_{}".format(_millis())

TABLE_ID = "get_table_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(DATASET_ID))

dataset = client.create_dataset(dataset)

to_delete.append(dataset)

table_ref = dataset.table(TABLE_ID)

table = bigquery.Table(table_ref, schema=SCHEMA)

table = client.create_table(table)

assert table_exists(client, table_ref)

assert not table_exists(client, dataset.table("i_dont_exist"))

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_manage_table_labels(client, to_delete):

dataset_id = "label_table_dataset_{}".format(_millis())

table_id = "label_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

client.create_dataset(dataset)

to_delete.append(dataset)

table = bigquery.Table(dataset.table(table_id), schema=SCHEMA)

table = client.create_table(table)

# [START bigquery_label_table]

# from google.cloud import bigquery

# client = bigquery.Client()

# table_ref = client.dataset('my_dataset').table('my_table')

# table = client.get_table(table_ref) # API request

assert table.labels == {}

labels = {"color": "green"}

table.labels = labels

table = client.update_table(table, ["labels"]) # API request

assert table.labels == labels

# [END bigquery_label_table]

# [START bigquery_get_table_labels]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset'

# table_id = 'my_table'

dataset_ref = client.dataset(dataset_id)

table_ref = dataset_ref.table(table_id)

table = client.get_table(table_ref) # API Request

# View table labels

print("Table ID: {}".format(table_id))

print("Labels:")

if table.labels:

for label, value in table.labels.items():

print("\t{}: {}".format(label, value))

else:

print("\tTable has no labels defined.")

# [END bigquery_get_table_labels]

assert table.labels == labels

# [START bigquery_delete_label_table]

# from google.cloud import bigquery

# client = bigquery.Client()

# table_ref = client.dataset('my_dataset').table('my_table')

# table = client.get_table(table_ref) # API request

# This example table starts with one label

assert table.labels == {"color": "green"}

# To delete a label from a table, set its value to None

table.labels["color"] = None

table = client.update_table(table, ["labels"]) # API request

assert table.labels == {}

# [END bigquery_delete_label_table]

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_update_table_description(client, to_delete):

"""Update a table's description."""

dataset_id = "update_table_description_dataset_{}".format(_millis())

table_id = "update_table_description_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

client.create_dataset(dataset)

to_delete.append(dataset)

table = bigquery.Table(dataset.table(table_id), schema=SCHEMA)

table.description = "Original description."

table = client.create_table(table)

# [START bigquery_update_table_description]

# from google.cloud import bigquery

# client = bigquery.Client()

# table_ref = client.dataset('my_dataset').table('my_table')

# table = client.get_table(table_ref) # API request

assert table.description == "Original description."

table.description = "Updated description."

table = client.update_table(table, ["description"]) # API request

assert table.description == "Updated description."

# [END bigquery_update_table_description]

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_update_table_expiration(client, to_delete):

"""Update a table's expiration time."""

dataset_id = "update_table_expiration_dataset_{}".format(_millis())

table_id = "update_table_expiration_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

client.create_dataset(dataset)

to_delete.append(dataset)

table = bigquery.Table(dataset.table(table_id), schema=SCHEMA)

table = client.create_table(table)

# [START bigquery_update_table_expiration]

import datetime

import pytz

# from google.cloud import bigquery

# client = bigquery.Client()

# table_ref = client.dataset('my_dataset').table('my_table')

# table = client.get_table(table_ref) # API request

assert table.expires is None

# set table to expire 5 days from now

expiration = datetime.datetime.now(pytz.utc) + datetime.timedelta(days=5)

table.expires = expiration

table = client.update_table(table, ["expires"]) # API request

# expiration is stored in milliseconds

margin = datetime.timedelta(microseconds=1000)

assert expiration - margin <= table.expires <= expiration + margin

# [END bigquery_update_table_expiration]

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_add_empty_column(client, to_delete):

"""Adds an empty column to an existing table."""

dataset_id = "add_empty_column_dataset_{}".format(_millis())

table_id = "add_empty_column_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

dataset = client.create_dataset(dataset)

to_delete.append(dataset)

table = bigquery.Table(dataset.table(table_id), schema=SCHEMA)

table = client.create_table(table)

# [START bigquery_add_empty_column]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset'

# table_id = 'my_table'

table_ref = client.dataset(dataset_id).table(table_id)

table = client.get_table(table_ref) # API request

original_schema = table.schema

new_schema = original_schema[:] # creates a copy of the schema

new_schema.append(bigquery.SchemaField("phone", "STRING"))

table.schema = new_schema

table = client.update_table(table, ["schema"]) # API request

assert len(table.schema) == len(original_schema) + 1 == len(new_schema)

# [END bigquery_add_empty_column]

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_relax_column(client, to_delete):

"""Updates a schema field from required to nullable."""

dataset_id = "relax_column_dataset_{}".format(_millis())

table_id = "relax_column_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

dataset = client.create_dataset(dataset)

to_delete.append(dataset)

# [START bigquery_relax_column]

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset'

# table_id = 'my_table'

original_schema = [

bigquery.SchemaField("full_name", "STRING", mode="REQUIRED"),

bigquery.SchemaField("age", "INTEGER", mode="REQUIRED"),

]

table_ref = client.dataset(dataset_id).table(table_id)

table = bigquery.Table(table_ref, schema=original_schema)

table = client.create_table(table)

assert all(field.mode == "REQUIRED" for field in table.schema)

# SchemaField properties cannot be edited after initialization.

# To make changes, construct new SchemaField objects.

relaxed_schema = [

bigquery.SchemaField("full_name", "STRING", mode="NULLABLE"),

bigquery.SchemaField("age", "INTEGER", mode="NULLABLE"),

]

table.schema = relaxed_schema

table = client.update_table(table, ["schema"])

assert all(field.mode == "NULLABLE" for field in table.schema)

# [END bigquery_relax_column]

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_update_table_cmek(client, to_delete):

"""Patch a table's metadata."""

dataset_id = "update_table_cmek_{}".format(_millis())

table_id = "update_table_cmek_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

client.create_dataset(dataset)

to_delete.append(dataset)

table = bigquery.Table(dataset.table(table_id))

original_kms_key_name = "projects/{}/locations/{}/keyRings/{}/cryptoKeys/{}".format(

"cloud-samples-tests", "us-central1", "test", "test"

)

table.encryption_configuration = bigquery.EncryptionConfiguration(

kms_key_name=original_kms_key_name

)

table = client.create_table(table)

# [START bigquery_update_table_cmek]

# from google.cloud import bigquery

# client = bigquery.Client()

assert table.encryption_configuration.kms_key_name == original_kms_key_name

# Set a new encryption key to use for the destination.

# TODO: Replace this key with a key you have created in KMS.

updated_kms_key_name = (

"projects/cloud-samples-tests/locations/us-central1/"

"keyRings/test/cryptoKeys/otherkey"

)

table.encryption_configuration = bigquery.EncryptionConfiguration(

kms_key_name=updated_kms_key_name

)

table = client.update_table(table, ["encryption_configuration"]) # API request

assert table.encryption_configuration.kms_key_name == updated_kms_key_name

assert original_kms_key_name != updated_kms_key_name

# [END bigquery_update_table_cmek]

def test_browse_table_data(client, to_delete, capsys):

"""Retreive selected row data from a table."""

# [START bigquery_browse_table]

# from google.cloud import bigquery

# client = bigquery.Client()

dataset_ref = client.dataset("samples", project="bigquery-public-data")

table_ref = dataset_ref.table("shakespeare")

table = client.get_table(table_ref) # API call

# Load all rows from a table

rows = client.list_rows(table)

assert len(list(rows)) == table.num_rows

# Load the first 10 rows

rows = client.list_rows(table, max_results=10)

assert len(list(rows)) == 10

# Specify selected fields to limit the results to certain columns

fields = table.schema[:2] # first two columns

rows = client.list_rows(table, selected_fields=fields, max_results=10)

assert len(rows.schema) == 2

assert len(list(rows)) == 10

# Use the start index to load an arbitrary portion of the table

rows = client.list_rows(table, start_index=10, max_results=10)

# Print row data in tabular format

format_string = "{!s:<16} " * len(rows.schema)

field_names = [field.name for field in rows.schema]

print(format_string.format(*field_names)) # prints column headers

for row in rows:

print(format_string.format(*row)) # prints row data

# [END bigquery_browse_table]

out, err = capsys.readouterr()

out = list(filter(bool, out.split("\n"))) # list of non-blank lines

assert len(out) == 11

@pytest.mark.skip(

reason=(

"update_table() is flaky "

"https://github.com/GoogleCloudPlatform/google-cloud-python/issues/5589"

)

)

def test_manage_views(client, to_delete):

project = client.project

source_dataset_id = "source_dataset_{}".format(_millis())

source_dataset_ref = client.dataset(source_dataset_id)

source_dataset = bigquery.Dataset(source_dataset_ref)

source_dataset = client.create_dataset(source_dataset)

to_delete.append(source_dataset)

job_config = bigquery.LoadJobConfig()

job_config.schema = [

bigquery.SchemaField("name", "STRING"),

bigquery.SchemaField("post_abbr", "STRING"),

]

job_config.skip_leading_rows = 1

uri = "gs://cloud-samples-data/bigquery/us-states/us-states.csv"

source_table_id = "us_states"

load_job = client.load_table_from_uri(

uri, source_dataset.table(source_table_id), job_config=job_config

)

load_job.result()

shared_dataset_id = "shared_dataset_{}".format(_millis())

shared_dataset_ref = client.dataset(shared_dataset_id)

shared_dataset = bigquery.Dataset(shared_dataset_ref)

shared_dataset = client.create_dataset(shared_dataset)

to_delete.append(shared_dataset)

# [START bigquery_create_view]

# from google.cloud import bigquery

# client = bigquery.Client()

# project = 'my-project'

# source_dataset_id = 'my_source_dataset'

# source_table_id = 'us_states'

# shared_dataset_ref = client.dataset('my_shared_dataset')

# This example shows how to create a shared view of a source table of

# US States. The source table contains all 50 states, while the view will

# contain only states with names starting with 'W'.

view_ref = shared_dataset_ref.table("my_shared_view")

view = bigquery.Table(view_ref)

sql_template = 'SELECT name, post_abbr FROM `{}.{}.{}` WHERE name LIKE "W%"'

view.view_query = sql_template.format(project, source_dataset_id, source_table_id)

view = client.create_table(view) # API request

print("Successfully created view at {}".format(view.full_table_id))

# [END bigquery_create_view]

# [START bigquery_update_view_query]

# from google.cloud import bigquery

# client = bigquery.Client()

# project = 'my-project'

# source_dataset_id = 'my_source_dataset'

# source_table_id = 'us_states'

# shared_dataset_ref = client.dataset('my_shared_dataset')

# This example shows how to update a shared view of a source table of

# US States. The view's query will be updated to contain only states with

# names starting with 'M'.

view_ref = shared_dataset_ref.table("my_shared_view")

view = bigquery.Table(view_ref)

sql_template = 'SELECT name, post_abbr FROM `{}.{}.{}` WHERE name LIKE "M%"'

view.view_query = sql_template.format(project, source_dataset_id, source_table_id)

view = client.update_table(view, ["view_query"]) # API request

# [END bigquery_update_view_query]

# [START bigquery_get_view]

# from google.cloud import bigquery

# client = bigquery.Client()

# shared_dataset_id = 'my_shared_dataset'

view_ref = client.dataset(shared_dataset_id).table("my_shared_view")

view = client.get_table(view_ref) # API Request

# Display view properties

print("View at {}".format(view.full_table_id))

print("View Query:\n{}".format(view.view_query))

# [END bigquery_get_view]

assert view.view_query is not None

analyst_group_email = "example-analyst-group@google.com"

# [START bigquery_grant_view_access]

# from google.cloud import bigquery

# client = bigquery.Client()

# Assign access controls to the dataset containing the view

# shared_dataset_id = 'my_shared_dataset'

# analyst_group_email = 'data_analysts@example.com'

shared_dataset = client.get_dataset(

client.dataset(shared_dataset_id)

) # API request

access_entries = shared_dataset.access_entries

access_entries.append(

bigquery.AccessEntry("READER", "groupByEmail", analyst_group_email)

)

shared_dataset.access_entries = access_entries

shared_dataset = client.update_dataset(

shared_dataset, ["access_entries"]

) # API request

# Authorize the view to access the source dataset

# project = 'my-project'

# source_dataset_id = 'my_source_dataset'

source_dataset = client.get_dataset(

client.dataset(source_dataset_id)

) # API request

view_reference = {

"projectId": project,

"datasetId": shared_dataset_id,

"tableId": "my_shared_view",

}

access_entries = source_dataset.access_entries

access_entries.append(bigquery.AccessEntry(None, "view", view_reference))

source_dataset.access_entries = access_entries

source_dataset = client.update_dataset(

source_dataset, ["access_entries"]

) # API request

# [END bigquery_grant_view_access]

def test_table_insert_rows(client, to_delete):

"""Insert / fetch table data."""

dataset_id = "table_insert_rows_dataset_{}".format(_millis())

table_id = "table_insert_rows_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

dataset = client.create_dataset(dataset)

dataset.location = "US"

to_delete.append(dataset)

table = bigquery.Table(dataset.table(table_id), schema=SCHEMA)

table = client.create_table(table)

# [START bigquery_table_insert_rows]

# TODO(developer): Uncomment the lines below and replace with your values.

# from google.cloud import bigquery

# client = bigquery.Client()

# dataset_id = 'my_dataset' # replace with your dataset ID

# For this sample, the table must already exist and have a defined schema

# table_id = 'my_table' # replace with your table ID

# table_ref = client.dataset(dataset_id).table(table_id)

# table = client.get_table(table_ref) # API request

rows_to_insert = [(u"Phred Phlyntstone", 32), (u"Wylma Phlyntstone", 29)]

errors = client.insert_rows(table, rows_to_insert) # API request

assert errors == []

# [END bigquery_table_insert_rows]

def test_load_table_from_file(client, to_delete):

"""Upload table data from a CSV file."""

dataset_id = "load_table_from_file_dataset_{}".format(_millis())

table_id = "load_table_from_file_table_{}".format(_millis())

dataset = bigquery.Dataset(client.dataset(dataset_id))

dataset.location = "US"

client.create_dataset(dataset)

to_delete.append(dataset)

snippets_dir = os.path.abspath(os.path.dirname(__file__))

filename = os.path.join(

snippets_dir, "..", "..", "bigquery", "tests", "data", "people.csv"

)

# [START bigquery_load_from_file]

# from google.cloud import bigquery

# client = bigquery.Client()

# filename = '/path/to/file.csv'

# dataset_id = 'my_dataset'

# table_id = 'my_table'

dataset_ref = client.dataset(dataset_id)

table_ref = dataset_ref.table(table_id)

job_config = bigquery.LoadJobConfig()

job_config.source_format = bigquery.SourceFormat.CSV

job_config.skip_leading_rows = 1

job_config.autodetect = True

with open(filename, "rb") as source_file:

job = client.load_table_from_file(

source_file,

table_ref,

location="US", # Must match the destination dataset location.

job_config=job_config,

) # API request

job.result() # Waits for table load to complete.

print("Loaded {} rows into {}:{}.".format(job.output_rows, dataset_id, table_id))

# [END bigquery_load_from_file]