I am trying to retrieve DDL objects from BigQuery but encountered a failure on line 18 with the code ddl_set = query_job.result(). I’ve assigned the necessary roles listed below to the service account.
- BigQuery Admin
- BigQuery Data Editor
- BigQuery Data Owner
- BigQuery Data Viewer
- BigQuery Job User
- BigQuery Metadata Viewer
- BigQuery User
- Owner
However, I received the following error message:
google.api_core.exceptions.Forbidden: 403 Access Denied: Table bq:{my_dataset}.INFORMATION_SCHEMA.TABLES: User does not have permission to query table bq:{my_dataset}.INFORMATION_SCHEMA.TABLES, or perhaps it does not exist in location US.
def get_execute_ddls():
creds=service_account.Credentials.from_service_account_file("/Users/stekavade/Desktop/bq_creds.json")
client = bigquery.Client(credentials=creds)
query = """
select schema_name from INFORMATION_SCHEMA.SCHEMATA
"""
query_job = client.query(query)
rows = query_job.result()
for row in rows:
schema = row.schema_name
print("Gathering ddl for tables in schema {}".format(schema))
query = """
SELECT table_name,replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(replace(ddl,'`',''),'INT64','INT'),'FLOAT64','FLOAT'),'BOOL','BOOLEAN'),'STRUCT','VARIANT'),'PARTITION BY','CLUSTER BY ('),';',');'),'CREATE TABLE ','CREATE TABLE if not exists '), "table INT,", '"table" INT,'),'_"table" INT,','_table INT,'),'ARRAY<STRING>','ARRAY'),'from','"from"'),'_"from"','_from'),'"from"_','from_'),'DATE(_PARTITIONTIME)','date(loaded_at)'),' OPTIONS(',', //'),'));',');'),'_at);','_at));'),'start ','"start" '),'_"start"','_start'),'order ','"order" '),'<',', //'),'_"order"','_order') as ddl
FROM `bq`.{}.INFORMATION_SCHEMA.TABLES where table_type='BASE TABLE'
"""
ddl_query = query.format(schema)
query_job = client.query(ddl_query)
ddl_set = query_job.result()
for row in ddl_set:
table_name = row.table_name
ddl = row.ddl
print("Running ddl for table {} in Snowflake".format(table_name))
use_schema = "use schema {}.{}".format("bq_db",schema)
with snowflake.connector.connect(
user='<snowflake_username>',
password='<snowflake_password>',
account='<snowflake_prod>',
warehouse='snwoflake_data_ingest',
database='bq_db',
role='bq_ingest_rl'
)as conn:
conn.cursor().execute(use_schema)
conn.cursor().execute(ddl)
print("Table {} created in bq_db.{} schema".format(table_name,schema))