Python Google Cloud Function缺失requested_run_time参数报错求助

背景流程 #

• 日志接收器（log sink）监控GA4每日表新增的特定日志
• 捕获日志后通过Pub/Sub消息触发Python Cloud Function
• Cloud Function执行多个调度查询

改动与报错 #

仅注释掉了transferids中的my_config_id4，自4月10日起函数失效，报错栈如下：

Traceback (most recent call last):
  File "/layers/google.python.pip/pip/lib/python3.12/site-packages/flask/app.py", line 2529, in wsgi_app
    response = self.full_dispatch_request()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/layers/google.python.pip/pip/lib/python3.12/site-packages/flask/app.py", line 1825, in full_dispatch_request
    rv = self.handle_user_exception(e)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/layers/google.python.pip/pip/lib/python3.12/site-packages/flask/app.py", line 1823, in full_dispatch_request
    rv = self.dispatch_request()
         ^^^^^^^^^^^^^^^^^^^^^^^
  File "/layers/google.python.pip/pip/lib/python3.12/site-packages/flask/app.py", line 1799, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/layers/google.python.pip/pip/lib/python3.12/site-packages/functions_framework/execution_id.py", line 106, in wrapper
    return view_function(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/layers/google.python.pip/pip/lib/python3.12/site-packages/functions_framework/__init__.py", line 142, in view_func
    return function(request._get_current_object())
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: runQuery() missing 1 required positional argument: 'requested_run_time'

环境信息：

• requirements.txt仅包含google-cloud-bigquery-datatransfer==1
• 基础镜像为Python 3.12（Ubuntu 22 Full）
• 函数最初由“Google Cloud Functions”部署，现在由个人邮箱账号部署

原函数代码：

import time
from google.protobuf.timestamp_pb2 import Timestamp
from google.cloud import bigquery_datatransfer_v1

def runQuery(parent, requested_run_time):
    client = bigquery_datatransfer_v1.DataTransferServiceClient()
    projectid = 'my_project_id'  # Enter your projectID here
    locationid = 'europe'
    transferids = [  #List of transfer IDs
                    'my_config_id1'   # Scheduled Query 1
                    ,'my_config_id2'  # Scheduled Query 2
                    ,'my_config_id3'  # Scheduled Query 3
                    #,'my_config_id4' # Scheduled Query 4 ## not needed anymore
                    ]  # List of transfer IDs

    for transferid in transferids:
        parent = client.location_transfer_config_path(projectid, locationid, transferid)
        start_time = bigquery_datatransfer_v1.types.Timestamp(seconds=int(time.time() + 10))
        response = client.start_manual_transfer_runs(parent, requested_run_time=start_time)

问题原因与修复 #

核心问题 #

Cloud Function被Pub/Sub触发时，只会传递一个request参数到入口函数，但原代码中runQuery定义了两个必填参数parent和requested_run_time，导致参数不匹配报错。同时函数内部已经重新生成了这两个参数的值，完全无需在函数定义中声明。

修复后的代码 #

修改函数定义以适配Pub/Sub触发规则，保留原有业务逻辑：

import time
from google.cloud import bigquery_datatransfer_v1

def runQuery(request):
    # 忽略Pub/Sub传递的request参数，无需使用
    client = bigquery_datatransfer_v1.DataTransferServiceClient()
    projectid = 'my_project_id'  # 替换为你的项目ID
    locationid = 'europe'
    transferids = [
                    'my_config_id1'   # 调度查询1
                    ,'my_config_id2'  # 调度查询2
                    ,'my_config_id3'  # 调度查询3
                    #,'my_config_id4' # 已废弃的调度查询4
                    ]

    for transferid in transferids:
        parent = client.location_transfer_config_path(projectid, locationid, transferid)
        start_time = bigquery_datatransfer_v1.types.Timestamp(seconds=int(time.time() + 10))
        response = client.start_manual_transfer_runs(parent, requested_run_time=start_time)
    
    # 返回合法响应，避免无响应警告
    return "Scheduled queries triggered successfully", 200

额外注意事项 #

1. 权限校验：个人邮箱部署的函数需要确保账号拥有bigquerydatatransfer.transfers.start权限，以及对应调度查询的操作权限。
2. 依赖升级：google-cloud-bigquery-datatransfer==1为旧版本，建议升级至最新稳定版（如google-cloud-bigquery-datatransfer==3.16.0），减少兼容性问题。
3. 返回值要求：Cloud Function必须返回HTTP响应，添加return语句可避免执行完成后的无响应警告。

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