使用Python在Pipedream向Google Cloud Pub/Sub发消息遇导入错误

问题描述 #

在Pipedream中使用Python代码向Google Cloud Pub/Sub发布消息时，遇到导入错误：

无法从'google.auth'导入名称'_cloud_sdk'

使用的代码片段：

import os
# pipedream add-package google-cloud-pubsub
from google.cloud import pubsub_v1
def handler(pd: "pipedream"):
  credentials_path =pd.steps["trigger"]["event"]
  os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = credentials_path
  publisher = pubsub_v1.PublisherClient()
  topic_path = 'projects/my-project-1909-363003/topics/my-topic'
  data = 'Msg from Pipedream'
  data = data.encode('utf-8')
  attributes = {
    'FName3': 'Acco',
  }
  future = publisher.publish(topic_path, data, **attributes)
  print(f'published message id {future.result()}')

错误原因 #

这个错误主要源于google-auth库版本与google-cloud-pubsub版本不兼容，或是Pipedream环境中存在依赖包冲突。另外代码里的凭证路径设置也有问题——直接把触发事件对象赋值给credentials_path，这并不是有效的凭证文件路径。

解决方案 #

1. 指定兼容的依赖版本 #

修改包安装命令，明确指定与google-cloud-pubsub兼容的google-auth版本，示例：

pipedream add-package google-cloud-pubsub google-auth==2.14.1

（版本号可根据google-cloud-pubsub官方文档调整，确保两者版本匹配）

2. 正确处理Google凭证 #

Pipedream触发事件的内容一般是JSON格式的凭证数据，而非文件路径。需要先把凭证内容写入临时文件，再设置环境变量指向该文件：

import os
import tempfile
# pipedream add-package google-cloud-pubsub google-auth==2.14.1
from google.cloud import pubsub_v1

def handler(pd: "pipedream"):
  # 从触发事件中获取凭证JSON数据
  credentials_data = pd.steps["trigger"]["event"]
  
  # 创建临时文件存储凭证
  with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as temp_file:
    temp_file.write(str(credentials_data))
    temp_cred_path = temp_file.name
  
  # 设置环境变量指向临时凭证文件
  os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = temp_cred_path
  
  try:
    publisher = pubsub_v1.PublisherClient()
    topic_path = 'projects/my-project-1909-363003/topics/my-topic'
    data = 'Msg from Pipedream'.encode('utf-8')
    attributes = {'FName3': 'Acco'}
    
    future = publisher.publish(topic_path, data, **attributes)
    print(f'published message id {future.result()}')
  finally:
    # 清理临时文件
    os.unlink(temp_cred_path)

3. 验证依赖安装 #

可以在Pipedream中添加调试步骤，打印已安装的目标包版本，确认依赖是否正确安装：

import pkg_resources
print([f"{pkg.key}=={pkg.version}" for pkg in pkg_resources.working_set if pkg.key in ['google-cloud-pubsub', 'google-auth']])

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