Cloud FunctionsPerformance Characteristics and Common Data Processing Patterns

# requirements:
# google-cloud-storage
# google-cloud-bigquery
#
import csv
import os
from tempfile import NamedTemporaryFile
from google.cloud import storage, bigquery

PROJECT_ID = "your-project-id"
BQ_DATASET = "your_dataset"
BQ_TABLE = "your_table"

storage_client = storage.Client()
bq_client = bigquery.Client(project=PROJECT_ID)

def gcs_csv_to_bigquery(event, context):
    """
    Cloud Function triggered by Cloud Storage when a CSV file is created.
    event contains keys like 'bucket' and 'name'.
    """
    bucket_name = event.get("bucket")
    object_name = event.get("name")
    if not bucket_name or not object_name:
        print("Missing bucket or name in event payload.")
        return

    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(object_name)

    # Download file to a temporary local file
    with NamedTemporaryFile(delete=False) as tmp:
        blob.download_to_file(tmp)
        tmp_path = tmp.name

    try:
        # Simple CSV parsing and streaming into BigQuery via load job
        table_ref = bq_client.dataset(BQ_DATASET).table(BQ_TABLE)

        job_config = bigquery.LoadJobConfig(
            source_format=bigquery.SourceFormat.CSV,
            skip_leading_rows=1,  # adjust if CSV has header
            autodetect=True       # or provide schema explicitly
        )

        with open(tmp_path, "rb") as source_file:
            load_job = bq_client.load_table_from_file(
                source_file, table_ref, job_config=job_config
            )
        load_job.result()  # wait for completion
        print(f"Loaded {object_name} into {BQ_DATASET}.{BQ_TABLE}")

    except Exception as e:
        print(f"Error loading {object_name} to BigQuery: {e}")

    finally:
        try:
            os.remove(tmp_path)
        except OSError:
            pass