Я пытаюсь использовать API Google Cloudiot из облачной функции Google, написанной на Python.Когда я делаю http-запросы к API Cloud IoT, я получаю 403 «запрещенную» ошибку. Я прошел аутентификацию, используя учетную запись службы с полными разрешениями.
import os from flask import json import base64
импорт firebase_admin из firebase_admin import db
импорт googleapiclient из googleapiclient обнаружение импорта из google.oauth2 import service_account
default_app = firebase_admin.initialize_app (options = {'databaseURL': '* 1009 1010* -v2-1.firebaseio.com / '})
def get_client (service_account_json): "" "Возвращает авторизованного клиента API, обнаруживая IoT API и создавая объект службы с использованием учетной записи службыучетные данные JSON. "" "api_scopes = ['https://www.googleapis.com/auth/cloud-platform',' https://www.googleapis.com/auth/cloudiot'] api_version = 'v1' discovery_api = 'https://cloudiot.googleapis.com/$discovery/rest' service_name =' cloudiotcore '
#credentials = service_account.Credentials.from_service_account_file(service_account_json)
credentials = service_account.Credentials.from_service_account_file(service_account_json)
scoped_credentials = credentials.with_scopes(api_scopes)
discovery_url = '{}?version={}'.format(
discovery_api, api_version)
print(credentials.service_account_email)
return discovery.build(
service_name,
api_version,
discoveryServiceUrl=discovery_url,
credentials=scoped_credentials,
cache_discovery=False)
def request_posted (событие, контекст):
print(os.environ)
#
client = get_client(os.environ.get('GOOGLE_APPLICATION_CREDENTIALS'))
"""Triggered by a change to a Firebase RTDB reference.
Args:
event (dict): Event payload.
context (google.cloud.functions.Context): Metadata for the event.
"""
print(client)
resources = context.resource.split('/')
request_id = resources[-1]
request_values = event['delta']
response_ref = db.reference("experience_channel/responses/" + str(request_id))
response_ref.update({"ack":True})
db.reference("experience_channel/experience_data")
device_id = "2593786060198708"
project_id = "lucid-iOS-v2-1"
cloud_region = "us-central1"
registry_id = "Lucid_IoT_Registry"
device_path = 'projects/{}/locations/{}/registries/{}/devices/{}'.format(project_id, cloud_region, registry_id, device_id)
print(device_path)
epoch_length = 20
message_contents = {
"experience_id":request_id,
"epoch_length":epoch_length,
"experience_length":request_values['length']
}
message_json = json.dumps(message_contents)
test_body = "test_body"
command_body = {
'binaryData': base64.urlsafe_b64encode(
message_json.encode('utf-8')).decode('ascii'),
'subfolder' : 'commands'
}
#client.projects().locations().registries().testIamPermissions(resource = )
print(client.projects(
).locations().registries(
).devices().list(parent = "projects/lucid-iOS-v2-1/locations/us-central1/registries/Lucid_IoT_Registry").execute())
resp = client.projects(
).locations().registries(
).devices().sendCommandToDevice(
name = device_path, body = command_body).execute()
print(resp)
return resp