google.api_core.exceptions.PermissionDenied, но служебная учетная запись имеет необходимые разрешения

Question

В настоящее время я использую python GCP API для создания очереди облачных задач. Мои коды изменены из примера кода, и лог c должен проверить, существует очередь или нет, если нет, создать новую очередь и поставить новую задачу в эту очередь. поэтому я использую try-исключения и импорт from google.api_core import exceptions для обработки ошибки. Но проблема сейчас говорит о том, что у моей учетной записи службы нет задачи облачного разрешения. Вот ошибка.

google.api_core.exceptions.PermissionDenied
google.api_core.exceptions.PermissionDenied: 403 The principal (user or service account) lacks IAM permission "cloudtasks.tasks.create" for the resource "projects/xxxx/locations/us-central1" (or the resource may not exist).

вот мой код.

@app.route('/train_model/<dataset_name>/<dataset_id>/', methods=["POST", "GET"])
def train_model(dataset_name,dataset_id):
    if request.method == 'POST':
        form = request.form
        model = form.get('model_name')
        date = form.get('date')
        datetime_object = datetime.strptime(date, '%Y-%m-%d %H:%M:%S')
        timezone = pytz.timezone('Asia/Hong_Kong')
        timezone_date_time_obj = timezone.localize(datetime_object)
        data=[dataset_id,model]
        payload = str(data).encode()

        # Create a client.

        url = "https://us-central1-xxx.cloudfunctions.net/create_csv"
        try:
            client = tasks_v2.CloudTasksClient.from_service_account_json(
                './xxxxx.json')

            url = "https://us-central1-xxxxxx.cloudfunctions.net/create_csv"
            location = 'us-central1'
            project = 'xxxxx'
            queue = 'testing1'
            parent = client.location_path(project, location)
            task = {
                "http_request": {
                    'http_method': 'POST',
                    'url': url,
                    'body': payload
                }}
            # set schedule time
            timestamp = timestamp_pb2.Timestamp()
            timestamp.FromDatetime(timezone_date_time_obj)
            task['schedule_time'] = timestamp
            response = client.create_task(parent, task)

        except exceptions.FailedPrecondition:
            location = 'us-central1'
            project = 397901391776
            # Create a client.
            client = tasks_v2.CloudTasksClient.from_service_account_json(
               "./xxxx.json")
            parent = client.location_path(project, location)
            queue = {"name": 'x'}
            queue.update(name="projects/xxxxx/locations/us-west2/queues/" + queue #the name of the queue from try.)
            response = client.create_queue(parent, queue)
            parent = client.queue_path(project, location, queue)
            task = {
            "http_request": {
                'http_method': 'POST',
                'url': url,
                'body':payload
            }}
            # set schedule time
            timestamp = timestamp_pb2.Timestamp()
            timestamp.FromDatetime(timezone_date_time_obj)
            task['schedule_time'] = timestamp
            response = client.create_task(parent, task)

        print(response)
        return redirect('/datasetinfo/{}/{}/'.format(dataset_name,dataset_id))

разрешение моей учетной записи службы

Answer 1

Я воспроизвел ваш сценарий, и мне удалось получить ту же проблему. Проблема не в аутентификации, а в том, что ресурс не существует.

Чтобы получить путь к ресурсу, вместо использования функции location_path следует использовать queue_path. Таким образом, переменная parent будет содержать имя очереди, и вызов create_task сможет найти ресурс.

Наконец, предоставление роли Editor для учетной записи службы может оказаться слишком сложным, вам следует ограничить доступ к минимуму жизнеспособных. Если этот код нужен только для создания задач, вам следует создать пользовательскую роль с необходимыми разрешениями, в данном случае cloudtasks.tasks.create.