如何从Django视图调用受服务账号限制的Google Cloud Function？

我来帮你理清两种可行的调用方式，同时解决你遇到的JSON格式错误问题：

方法一：直接发送HTTP请求（推荐用于HTTP触发的函数） #

因为你的Cloud Function是通过HTTP触发的，直接用requests库配合服务账号生成的ID Token授权是最直观的方式，和你在控制台测试HTTP触发的逻辑完全一致：

1. 先确保你的服务账号拥有cloudfunctions.invoker权限（否则会触发权限不足错误）
2. 代码示例：

from google.oauth2 import service_account
import google.auth.transport.requests
import requests
import json

# 替换为你的配置信息
SERVICE_ACCOUNT_FILE = 'credentials/credentials.json'
CLOUD_FUNCTION_URL = 'https://<REGION>-<PROJECT_ID>.cloudfunctions.net/<FUNCTION_NAME>'
REQUEST_PAYLOAD = {"user_id": 123, "action": "process_data"}

# 生成针对该Cloud Function的ID Token（受众必须是函数的HTTP URL）
credentials = service_account.IDTokenCredentials.from_service_account_file(
    SERVICE_ACCOUNT_FILE,
    target_audience=CLOUD_FUNCTION_URL
)

# 获取有效Token
request = google.auth.transport.requests.Request()
credentials.refresh(request)

# 发送POST请求
response = requests.post(
    CLOUD_FUNCTION_URL,
    headers={
        'Authorization': f'Bearer {credentials.token}',
        'Content-Type': 'application/json'
    },
    data=json.dumps(REQUEST_PAYLOAD)
)

# 处理响应
if response.ok:
    print("调用成功:", response.json())
else:
    print(f"调用失败，状态码: {response.status_code}，错误信息: {response.text}")

这种方式下，你的Cloud Function内部可以直接用request.get_json()拿到你发送的JSON数据，不需要额外解析。

方法二：修复你用googleapiclient调用的错误 #

你之前尝试用Cloud Functions管理API的call方法，但遇到了JSON格式错误，问题出在请求体的结构上：

call方法要求请求体是CallFunctionRequest格式，你的业务数据必须放在data字段中，而且不能提前把整个body转成JSON字符串（googleapiclient会自动处理序列化）。正确代码如下：

from google.oauth2 import service_account
import googleapiclient.discovery
import json

SCOPES = ['https://www.googleapis.com/auth/cloud-platform']
SERVICE_ACCOUNT_FILE = 'credentials/credentials.json'
credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_FILE, scopes=SCOPES)

cloudfunction = googleapiclient.discovery.build('cloudfunctions', 'v1', credentials=credentials)

# 替换为你的函数完整路径（格式必须严格遵循：projects/{project_id}/locations/{location_id}/functions/{function_id}）
FUNCTION_PATH = 'projects/<PROJECT_ID>/locations/<REGION>/functions/<FUNCTION_NAME>'

# 你的业务数据
payload = {"key": "value", "another_key": "another_value"}

# 构造正确的请求体：把业务数据转成JSON字符串后放在data字段中
call_request_body = {
    "data": json.dumps(payload)
}

# 调用函数，注意body传字典，不是字符串
response = cloudfunction.projects().locations().functions().call(
    name=FUNCTION_PATH,
    body=call_request_body
).execute()

print("调用结果:", response)

需要注意：用这种方式调用时，你的Cloud Function内部需要额外解析data字段的字符串，比如：

# Cloud Function内部代码示例
def hello_world(request):
    request_data = json.loads(request.get_json()['data'])
    # 处理数据逻辑
    return json.dumps({"status": "success", "received": request_data})

这种方式更适合非HTTP触发的函数，或者需要通过管理API触发函数的场景，对于HTTP触发的函数，还是推荐第一种方法。

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