Python Cloud Function调用API写入BigQuery无数据无报错问题

我正在开发首个Cloud Function，功能为从API拉取数据、转换为DataFrame后写入BigQuery。已配置HTTP触发器，将validate_http设为入口函数。函数显示执行成功，但未向BigQuery写入任何数据。

本地测试时，get_api_data()函数可正常运行并写入BigQuery，但部署到Cloud Functions后无法触发数据写入，且无报错信息。

代码实现 #

import pandas as pd
import json
import requests
from pandas.io import gbq
import pandas_gbq
import gcsfs

# function 1: Responding and validating any HTTP request
def validate_http(request):
    request.json = request.get_json()
    
    if request.args:
        get_api_data()
        return f'Data pull complete'
    
    elif request_json:  # 此处存在变量名错误
        get_api_data()
        return f'Data pull complete'
    
    else:
        get_api_data()
        return f'Data pull complete'

# function 2: Get data and transform
def get_api_data():
    import pandas as pd
    import requests
    import json

    # Setting up variables with tokens
    base_url = "https://"
    token= "&token="
    token2= "&token="
    fields = "&fields=date,id,shippingAddress,items"
    date_filter = "&filter=date in '2022-01-22'"
    data_limit = "&limit=99999999"

    # Performing API call on request with variables
    def main_requests(base_url,token,fields,date_filter,data_limit):
        req = requests.get(base_url + token + fields +date_filter + data_limit)
        return req.json()

    # Making API Call and storing in data
    data = main_requests(base_url,token,fields,date_filter,data_limit)

    # transforming the data
    df = pd.json_normalize(data['orders']).explode('items').reset_index(drop=True)
    items = df['items'].agg(pd.Series)[['id','itemNumber','colorNumber', 'amount', 'size','quantity', 'quantityReturned']]
    df = df.drop(columns=[ 'items', 'shippingAddress.id', 'shippingAddress.housenumber', 'shippingAddress.housenumberExtension', 'shippingAddress.address2','shippingAddress.name','shippingAddress.companyName','shippingAddress.street', 'shippingAddress.postalcode', 'shippingAddress.city', 'shippingAddress.county', 'shippingAddress.countryId', 'shippingAddress.email', 'shippingAddress.phone'])
    df = df.rename(columns=
         {'date' : 'Date',
          'shippingAddress.countryIso' : 'Country',
          'id' : 'order_id'})

    df = pd.concat([df, items], axis=1, join='inner')      
  
    # Push data function
    bq_load('Return_data_api', df)  # 原代码此处缩进错误，需修正

# function 3: Convert to bigquery table
def bq_load(key, value):
    project_name = '375215'
    dataset_name = 'Returns'
    table_name = key
    
    value.to_gbq(destination_table='{}.{}'.format(dataset_name, table_name), project_id=project_name, if_exists='replace')

排查与解决建议 #

• 修正变量名错误：validate_http函数中使用了未定义的request_json，应改为request.json。该错误可能导致代码逻辑异常，虽函数返回成功，但实际未正确执行后续流程。
• 修复缩进问题：get_api_data函数中调用bq_load的代码缩进错误（原代码中这一行缩进层级不足），导致写入BigQuery的逻辑从未被执行，这是本地运行正常但部署后失效的核心原因之一。
• 添加日志排查：在关键节点添加日志打印，确认数据流转和函数执行情况。例如：

  import logging
  
  def get_api_data():
      # ... 原有代码 ...
      df = pd.concat([df, items], axis=1, join='inner')
      logging.info(f"转换后数据行数：{len(df)}")
      bq_load('Return_data_api', df)
  
  def bq_load(key, value):
      logging.info(f"开始写入BigQuery表：{key}")
      # ... 原有代码 ...
      value.to_gbq(...)
      logging.info("BigQuery写入完成")
  

  随后通过Cloud Functions的日志界面查看执行细节，确认是否有隐藏异常。
• 验证服务账号权限：确保Cloud Function使用的服务账号拥有BigQuery数据集的bigquery.dataEditor或更高权限，避免因权限不足导致写入失败。
• 统一依赖版本：在requirements.txt中明确指定依赖包版本，与本地测试环境保持一致，例如：

  pandas==2.1.4
  pandas-gbq==0.19.2
  requests==2.31.0
  

内容的提问来源于stack exchange，提问作者quasimus