django - 未能在 Google Cloud Tasks 上执行的每个任务

Question

我需要在 Django 应用中运行一些异步任务，于是我开始研究 Google Cloud Tasks。我想我已经遵循了所有的说明——以及我能想到的每一种可能的变化，到目前为止都没有成功。

问题是所有创建的任务都进入了队列，但无法执行。控制台和日志仅报告 http 代码 301(永久重定向)。为简单起见，我将相同的代码部署到 App Engine(标准)的两个服务中，并将任务请求仅路由到其中一个。

看起来代码本身运行良好。当我转到“https://[proj].appspot.com/api/v1/tasks”时，例程执行得很好，并且根据 DevTools/Network 没有重定向。当 Cloud Tasks 尝试调用“/api/v1/tasks”时，它每次都失败。

如果有人可以查看下面的代码并指出可能导致此失败的原因，我将不胜感激。

谢谢你。

#--------------------------------
# [proj]/.../urls.py
#--------------------------------
from [proj].api import tasks

urlpatterns += [
    # tasks api
    path('api/v1/tasks', tasks, name='tasks'),
]

#--------------------------------
#   [proj]/api.py:
#--------------------------------
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def tasks(request):
    print('Start api')
    payload = request.body.decode("utf-8")
    print (payload)
    print('End api')
    return HttpResponse('OK')

#--------------------------------
# [proj]/views/manut.py
#--------------------------------
from django.views.generic import View
from django.shortcuts import redirect
from [proj].tasks import TasksCreate

class ManutView(View):
    template_name = '[proj]/manut.html'

    def post(self, request, *args, **kwargs):
        relative_url = '/api/v1/tasks'
        testa_task = TasksCreate()
        resp = testa_task.send_task(
            url=relative_url,
            schedule_time=5,
            payload={'task_type': 1, 'id': 21}
        )
        print(resp)
        return redirect(request.META['HTTP_REFERER'])

#--------------------------------
# [proj]/tasks/tasks.py:
#--------------------------------
from django.conf import settings
from google.cloud import tasks_v2
from google.protobuf import timestamp_pb2
from typing import Dict, Optional, Union
import json
import time

class TasksCreate:

    def send_task(self,
        url: str,
        payload: Optional[Union[str, Dict]] = None,
        schedule_time: Optional[int] = None,    # in seconds
        name: Optional[str] = None,
        ) -> None:

        client = tasks_v2.CloudTasksClient()
        parent = client.queue_path(
            settings.GCP_PROJECT,
            settings.GCP_LOCATION,
            settings.GCP_QUEUE,
        )

        # App Engine task:
        task = {
            'app_engine_http_request': {  # Specify the type of request.
                'http_method': 'POST',
                'relative_uri': url,
                'app_engine_routing': {'service': 'tasks'}
            }
        }

        if name:
            task['name'] = name
        if isinstance(payload, dict):
            payload = json.dumps(payload)
        if payload is not None:
            converted_payload = payload.encode()
            # task['http_request']['body'] = converted_payload
            task['app_engine_http_request']['body'] = converted_payload
        if schedule_time is not None:
            now = time.time() + schedule_time
            seconds = int(now)
            nanos = int((now - seconds) * 10 ** 9)
            # Create Timestamp protobuf.
            timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos)
            # Add the timestamp to the tasks.
            task['schedule_time'] = timestamp

        resp = client.create_task(parent, task)

        return resp

# --------------------------------
# [proj]/dispatch.yaml:
# --------------------------------
dispatch:
  - url: "*/api/v1/tasks"
    service: tasks

  - url: "*/api/v1/tasks/"
    service: tasks

  - url: "*appspot.com/*"
    service: default

#--------------------------------
# [proj]/app.yaml & tasks.yaml:
#--------------------------------
runtime: python37

instance_class: F1

automatic_scaling:
  max_instances: 2

service: default

#handlers:
#- url: .*
#  secure: always
#  redirect_http_response_code: 301
#  script: auto

entrypoint: gunicorn -b :$PORT --chdir src server.wsgi

env_variables:
...

更新:

以下是执行的日志:

{
 insertId: "1lfs38fa9"  
 jsonPayload: {
  @type: "type.googleapis.com/google.cloud.tasks.logging.v1.TaskActivityLog"   
  attemptResponseLog: {
   attemptDuration: "0.008005s"    
   dispatchCount: "5"    
   maxAttempts: 0    
   responseCount: "5"    
   retryTime: "2020-03-09T21:50:33.557783Z"    
   scheduleTime: "2020-03-09T21:50:23.548409Z"    
   status: "UNAVAILABLE"    
   targetAddress: "POST /api/v1/tasks"    
   targetType: "APP_ENGINE_HTTP"    
  }
  task: "projects/[proj]/locations/us-central1/queues/tectaq/tasks/09687434589619534431"   
 }
 logName: "projects/[proj]/logs/cloudtasks.googleapis.com%2Ftask_operations_log"  
 receiveTimestamp: "2020-03-09T21:50:24.375681687Z"  
 resource: {
  labels: {
   project_id: "[proj]"    
   queue_id: "tectaq"    
   target_type: "APP_ENGINE_HTTP"    
  }
  type: "cloud_tasks_queue"   
 }
 severity: "ERROR"  
 timestamp: "2020-03-09T21:50:23.557842532Z"  
}

Answer 1

最后我可以让 Cloud Tasks 工作，但只使用 http_request 类型(使用绝对 url)。当任务被定义为 app_engine_http_request(相对 url)时，我无法让它们运行。

我已经用 POST 尝试了 http_request 类型，但那是在我免除 api 函数之前检查 csrf token 之前，这导致了错误 Forbidden (Referer checking failed - no Referer.): /api/v1/tasks ，我未能连接到 csrf 遗漏。

如果将来有人偶然发现这个问题，并找到一种方法让 app_engine_http_request 使用 Django 在 Cloud Tasks 上工作，我仍然非常想知道解决方案。