我如何将 Celery 任务转换为 Google Cloud 任务 GCP (Django)

Question

我在本地主机上使用 Celery + Redis 和 Django Rest API 来运行分类任务，如何从 Axios 帖子获取数据。现在我正在尝试将其部署到谷歌云，并且我没有找到在App Engine上运行Redis和Celery的明确方法，所以我听说了Google任务队列，但我没有找到将其添加到视图和触发器的方法当视图被调用时，我如何创建一个函数来调用我在 celery 上的这个谷歌云任务，或者只是知道如何做到这一点，这些是我的代码：

from celery import shared_task
from celery_progress.backend import ProgressRecorder

from snakeimage.models import Prediction,UploadedSnake,SnakeClass
from snakeimage.classification_codes.classification_codes.prediction_func import predict_classes

#import json
#import time

#from django.conf import settings

#from google.cloud import tasks_v2beta3
#from google.protobuf import timestamp_pb2

@shared_task(bind=True)
 def image_progress(self,image_path, X, Y, metadata,image_id):
 progress_recorder = ProgressRecorder(self,)
 predictions = predict_classes(image_path, X, Y, metadata)
 print(predictions)
   for prediction in predictions:
      print(prediction[0])
      image = UploadedSnake.objects.get(id=image_id)
      class_name = SnakeClass.objects.get(index=(prediction[0]+1))
      print('>>>>>>>>>>>>>>>>>>>>>',prediction[1])
     
 Prediction.objects.create(image=image,class_name=class_name,predict_percent=prediction[1])
 progress_recorder.set_progress( 1, 3, description='Prediction Result Status')
 return True

我用 a 来称呼它：

task = image_progress.delay(image_path=image_path, X=X, Y=Y, metadata=0, image_id=image_id)

谢谢您的帮助 。

[编辑1]

很抱歉迟到的回复，我可以让它在本地与 Django-cloud-task 一起工作，但在临时服务器中它不起作用，当我尝试从我的机​​器远程连接到我创建的 Google Cloud 任务队列时，我收到此错误它尝试发送 7 次： googleapiclient.errors.HttpError: <HttpError 403 when requesting https://cloudtasks.googleapis.com/v2beta3/projects/%7Bdeployment-test%7D/locations/europe-west6/queues/ default/tasks?alt=json返回“资源项目 {deployment-test} 的权限被拒绝。”。详细信息：“[{'@type': 'type.googleapis.com/google.rpc.Help', 'links': [{'description': 'Google 开发者控制台 API 密钥', 'url': 'https:// /console.developers.google.com/project/{deployment-test}/apiui/credential'}]}, {'@type': 'type.googleapis.com/google.rpc.ErrorInfo', '原因': ' CONSUMER_INVALID', 'domain': 'googleapis.com', 'metadata': {'consumer': 'projects/{deployment-test}', 'service': 'cloudtasks.googleapis.com'}}]">

我做了 github 中提到的所有操作，他们没有添加一些 api 密钥来进行身份验证的选项，所以有人知道如何解决这个问题，这些是我的代码：设置：

 DJANGO_CLOUD_TASKS_EXECUTE_LOCALLY = False
 # If False, running `.execute()` on remote task will simply log the 
 task data instead of adding it to
 # the queue. Useful for debugging. Default: True
 DJANGO_CLOUD_TASKS_BLOCK_REMOTE_TASKS = True

PROJECT_NAME = "project"
QUEUE_REGION = "region"
QUEUE_NAME = "queue"

DJANGO_CLOUD_TASKS_HANDLER_SECRET = 'random secret key'

DJANGO_CLOUD_TASKS = {
                     'project_location_name': 'projects/{project}/locations/region',
                     'task_handler_root_url': '/_tasks/',
                 }

任务.py:

from celery import shared_task
#from celery_progress.backend import ProgressRecorder

from snakeimage.models import Prediction,UploadedSnake,SnakeClass
from 
snakeimage.classification_codes.classification_codes.prediction_func 
import predict_classes
#from AINature.settings import DJANGO_HANDLER_SECRET

import json
import time

from django.conf import settings

from google.cloud import tasks_v2beta3
from google.protobuf import timestamp_pb2

from django_cloud_tasks.decorators import task

@task(queue='default')
def example_task(request, p1, p2):
    print(p1, p2)
    print("lezgooow >>>>>>>>>>>>>>>"+p1)
    print(request.task_id)

 def prediction_task(request,image_path, X, Y, metadata,image_id):
    print("what is going ooon")
    #progress_recorder = ProgressRecorder(self, )
    predictions = predict_classes(image_path, X, Y, metadata)
    print(predictions)
    for prediction in predictions:
        print(prediction[0])
        image = UploadedSnake.objects.get(id=image_id)
        class_name = SnakeClass.objects.get(index=(prediction[0] + 1))
        print('>>>>>>>>>>>>>>>>>>>>>', prediction[1])
        Prediction.objects.create(image=image, class_name=class_name, 

Predict_percent=预测1 ) #progress_recorder.set_progress(1, 3, description='预测结果状态') return True

当我设置 DJANGO_CLOUD_TASKS_EXECUTE_LOCALLY = True 时，所有操作都正常运行，但是当我将其关闭时，它会抛出我提到的错误。这是 Django-cloud-task github 的链接：这个链接

Answer 1

Celery 和 Cloud Task 都是任务队列，但实现不同，因此没有直接的方法将 Celery 转换为 Cloud Task 逻辑。这意味着如果您仅使用云任务服务会更容易。我建议在迁移之前研究Cloud Task Client 库。GitHub链接中有一些示例可以帮助您入门。

如果您想仍然使用 Celery，您将需要弄清楚如何通过 HTTP 请求触发工作程序，因为 App Engine Standard 仅接受 HTTP 请求。

您还可以使用其他服务选项，Compute Engine和App Engine Flex，您可以在其中实施您想要的设置。