Notes

import logging

from airflow import AirflowException
from google.cloud import bigquery

logger = logging.getLogger(__name__)


def util_bq_operation(*ds, **kwargs):
try:
# Construct a BigQuery client object.
client = bigquery.Client()

# replace 2022, with 2023 and change where condition to have 2022 tables
sql = """SELECT table_name,replace(replace(ddl,"2022","2023"),"CREATE TABLE","CREATE TABLE IF NOT EXISTS")ddl FROM
alta_sys_us_bq_product_price_bronze.INFORMATION_SCHEMA.TABLES
where table_name like '%2022'"""

# Start the query, passing in the extra configuration.
query_job = client.query(sql) # Make an API request.
response_bq = query_job.result() # Wait for the job to complete.
for row in response_bq:
upcoming_year_ddl = row.ddl
this_year_table = row.table_name
create_upcoming_year_tbl = client.query(upcoming_year_ddl)
create_upcoming_year_tbl.result()
logger.info(f"Upcoming year table for this year table"
f" {this_year_table} "
f"has been created successfully")

except Exception as e:
raise AirflowException(
'Error while creating upcoming year tables in BQ'.format(
str(e)))