Skip to main content

snippets_py

Integration​

New integration from credential
big_query_client = yepcode.integration.google_big_query("credential-slug")
New integration from plain authentication data
from google.cloud.bigquery.client import Client
from google.oauth2.service_account import Credentials

project_id = "yepcode"
credentialsDict = {
    "type": "service_account",
    "project_id": "yepcode",
    "private_key_id": "XXXXX",
    "private_key": "-----BEGIN PRIVATE KEY-----\nx\n-----END PRIVATE KEY-----",
    "client_email": "yepcode@example.org",
    "client_id": "1234567890",
    "auth_uri": "https://example.org",
    "token_uri": "https://example.org",
    "auth_provider_x509_cert_url": "https://example.org",
    "client_x509_cert_url": "https://example.org",
}
credentials = Credentials.from_service_account_info(credentialsDict)
big_query_client = Client(project=project_id, credentials=credentials)

Create a Dataset​

Create a dataset
from google.cloud import bigquery

dataset_name = "dataset-name"
dataset_id = f"{big_query_client.project}.{dataset_name}"
dataset = bigquery.Dataset(dataset_id)
dataset.location = "dataset-location"

dataset = big_query_client.create_dataset(dataset)

List Datasets​

List datasets
datasets = list(big_query_client.list_datasets())
for dataset in datasets:
  print(dataset.dataset_id)

Delete a Dataset​

Delete a dataset
dataset_name = "dataset-name"
dataset_id = f"{big_query_client.project}.{dataset_name}"
big_query_client.delete_dataset(dataset_id, delete_contents=True, not_found_ok=True)

Query to a Dataset​

Query to a dataset
dataset_id = "dataset-id"
table_id = "table-id"
query=f"SELECT * FROM `{big_query_client.project}.{dataset_id}.{table_id}`"

query_job = big_query_client.query(query, location="dataset-location")
rows = query_job.result()

for row in rows:
    print(row)

Insert Data to a Table​

Insert data to a table
rows_to_insert = [("Foo", 1), ("BAR", 2)]
table_ref = big_query_client.dataset("dataset-id").table("table-id")
table = big_query_client.get_table(table_ref)
errors = big_query_client.insert_rows(table, rows_to_insert)

if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))