如何通过Python客户端获取BigQuery项目中所有计划查询的名称

前提准备 #

• 安装依赖库：

  pip install google-cloud-bigquery-datatransfer
  

• 完成身份认证：
  • 本地开发可执行gcloud auth application-default login
  • 生产环境设置GOOGLE_APPLICATION_CREDENTIALS环境变量，指向服务账号JSON密钥文件

核心代码实现 #

计划查询在BigQuery中属于Data Transfer Service的一种配置类型，通过筛选data_source_id为scheduled_query的传输配置，即可获取所有计划查询：

from google.cloud import bigquery_datatransfer

def list_scheduled_queries(project_id):
    # 初始化Data Transfer客户端
    client = bigquery_datatransfer.DataTransferServiceClient()
    # 构造项目父资源路径
    parent = f"projects/{project_id}"

    # 列出项目下所有数据传输配置
    transfer_configs = client.list_transfer_configs(parent=parent)

    # 筛选出计划查询类型的配置
    scheduled_queries = [
        config for config in transfer_configs
        if config.data_source_id == "scheduled_query"
    ]

    # 输出计划查询信息
    for idx, query in enumerate(scheduled_queries, 1):
        print(f"[{idx}] 计划查询名称: {query.display_name}")
        print(f"    资源ID: {query.name}\n")

if __name__ == "__main__":
    # 替换为你的GCP项目ID
    YOUR_PROJECT_ID = "your-gcp-project-id"
    list_scheduled_queries(YOUR_PROJECT_ID)

关键说明 #

• query.display_name：对应你在BigQuery控制台创建计划查询时设置的名称
• query.name：计划查询的完整资源路径（格式为projects/{project_id}/locations/{location}/transferConfigs/{config_id}）
• 权限要求：执行代码的账号需要拥有bigquery.transfers.list权限，可通过分配BigQuery Data Transfer Viewer或BigQuery Data Transfer Editor角色实现

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