def read_event_data()

in bigquery-jobs-notifier/run/app.py [0:0]

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

def read_event_data(cloud_event):

    # Assume custom event by default
    event_data = cloud_event.data

    protoPayload = event_data['protoPayload']
    principalEmail = protoPayload['authenticationInfo']['principalEmail']
    job = protoPayload['serviceData']['jobCompletedEvent']['job']
    jobId = job['jobName']['jobId']
    jobStatistics = job['jobStatistics']
    createTime = jobStatistics['createTime']
    totalBilledBytes = 0
    if 'totalBilledBytes' in jobStatistics:
        totalBilledBytes = float(jobStatistics['totalBilledBytes'])
    query = job['jobConfiguration']['query']['query']
    aboveLimit = totalBilledBytes > 1000000000

    message = f"""
The following BigQuery job completed

principalEmail: {principalEmail}
jobId: {jobId}
createTime: {createTime}
query: {query}
totalBilledBytes: {totalBilledBytes}, above 1GB? {aboveLimit}"""

    app.logger.info(message)

    return aboveLimit, message