在 Cloud Function 中调用 Google API 并进行身份验证

Question

我正在尝试从云函数中调用 Dataform API，但是我提供的身份令牌返回时带有

'Request had invalid authentication credentials. Expected OAuth 2 access token, login cookie or other valid authentication credential.'

根据文档，运行云功能的SA只需要

dataform.workflowInvocations.query

权限，我已授予该权限并确认这在使用Python Dataform库时有效（出于技术原因，我需要调用REST API端点Python库不提供）。

这是我用来通过授权令牌发出请求的代码，这是我从 Google 获得的代码here，但是使用

requests

库来进行 API 调用。

import requests
import google.auth.transport.requests
import google.oauth2.id_token

def get_workflow_invocation_actions(environment, git_repo, workflow_invocation_id):
    parent = f"projects/{environment}/locations/europe-west2/repositories/{git_repo}"
    name = f"{parent}/workflowInvocations/{workflow_invocation_id}"

    # The API endpoint
    url = f"https://dataform.googleapis.com/v1beta1/{name}:query"

    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, url)

    # A GET request to the API
    response = requests.get(url, headers={'Authorization': f'Bearer {id_token}'})

    # Print the response
    print(response.json())

id_token

确实填充了一个看起来有效的令牌，所以我不确定是什么阻止它通过 REST API 进行身份验证。

Answer 1

在深入研究所有可能的文档后，我自己解决了这个问题，事实证明这与我正在使用的 API 语法有关。它使用 gRPC Transcoding Syntax，这需要使用大括号作为您调用的 URL 端点的一部分。我犯了一个简单的错误，认为我调用的端点的大括号是文档中表示 URL 的这一部分需要由用户填充的方式，因此当我填充它时我忽略了它们。 所以代码现在看起来像这样（注意现在包含的 URL 变量的附加大括号）：

import requests
import google.auth.transport.requests
import google.oauth2.id_token

def get_workflow_invocation_actions(environment, git_repo, workflow_invocation_id):
    parent = f"projects/{environment}/locations/europe-west2/repositories/{git_repo}"
    name = f"{parent}/workflowInvocations/{workflow_invocation_id}"

    # The API endpoint
    url = f"https://dataform.googleapis.com/v1beta1/{{{name}}}:query"

    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, url)

    # A GET request to the API
    response = requests.get(url, headers={'Authorization': f'Bearer {id_token}'})

    # Print the response
    print(response.json())