def insert_row_into_bigquery()

in webhooks-with-cloud-run/Monolith/main.py [0:0]

• 24 lines of code
• 2 McCabe index (conditional complexity)

def insert_row_into_bigquery(event):
    from google.cloud import bigquery

    # Set up bigquery instance
    client = bigquery.Client()
    dataset_id = os.environ.get("DATASET")
    table_id = os.environ.get("TABLE")
    table_ref = client.dataset(dataset_id).table(table_id)
    table = client.get_table(table_ref)

    # Insert row
    row_to_insert = [
        (
            event["issue"]["title"],
            event["action"],
            event["issue"]["html_url"],
            time.time(),
        )
    ]
    bq_errors = client.insert_rows(table, row_to_insert)

    # If errors, log to Stackdriver
    if bq_errors:
        entry = {
            "severity": "WARNING",
            "msg": "Row not inserted.",
            "errors": bq_errors,
            "row": row_to_insert,
        }
        print(json.dumps(entry))