我试图知道云函数URL是否总是响应,了解URL的状态并在GCP监控中显示它是很重要的。
是否有可能知道它是否是活动的。如果有可能的话,有人能帮我处理示例代码吗?
我试着像下面这样,
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "sa.json"
cf_url = f'https://{region}-{self.cf_project_id}.cloudfunctions.net/{self.cf_name}'
var1=requests.get(cf_url)
print(var1.status_code)
我期望这个get调用应该给我状态代码200,以知道call已经启动并正常。但我得到了403级。
该服务帐户具有云功能查看器访问权限!
发布于 2022-07-06 17:50:09
这是意料之中的。您获得HTTP 403
是因为您的请求没有经过身份验证。
将GOOGLE_APPLICATION_CREDENTIALS
变量设置为服务帐户不会自动设置身份验证头。
此外,您需要的角色是Cloud Functions Invoker
,而不是Cloud Function Viewer
。Cloud Function Viewer
用于查看函数,而不是触发它们。
您可以像在这个answer中看到的那样尝试这一点。
from google.oauth2 import service_account
from google.auth.transport.requests import AuthorizedSession
url = 'https://test-123456.cloudfunctions.net/my-cloud-function'
creds = service_account.IDTokenCredentials.from_service_account_file(
'/path/to/service-account-credentials.json', target_audience=url)
authed_session = AuthorizedSession(creds)
# make authenticated request and print the response, status_code
resp = authed_session.get(url)
print(resp.status_code)
print(resp.text)
或由Jonh here提供的代码
import json
import base64
import requests
import google.auth.transport.requests
from google.oauth2.service_account import IDTokenCredentials
# The service account JSON key file to use to create the Identity Token
sa_filename = 'service-account.json'
# Endpoint to call
endpoint = 'https://us-east1-replace_with_project_id.cloudfunctions.net/main'
# The audience that this ID token is intended for (example Google Cloud Functions service URL)
aud = 'https://us-east1-replace_with_project_id.cloudfunctions.net/main'
def invoke_endpoint(url, id_token):
headers = {'Authorization': 'Bearer ' + id_token}
r = requests.get(url, headers=headers)
if r.status_code != 200:
print('Calling endpoint failed')
print('HTTP Status Code:', r.status_code)
print(r.content)
return None
return r.content.decode('utf-8')
if __name__ == '__main__':
credentials = IDTokenCredentials.from_service_account_file(
sa_filename,
target_audience=aud)
request = google.auth.transport.requests.Request()
credentials.refresh(request)
# This is debug code to show how to decode Identity Token
# print('Decoded Identity Token:')
# print_jwt(credentials.token.encode())
response = invoke_endpoint(endpoint, credentials.token)
if response is not None:
print(response)
https://stackoverflow.com/questions/72872915
复制相似问题