删除实体的任务队列失败 - python - flask - google appengine

Question

我不知道我在做什么。

这就是我的 google appengine 任务队列无法正常工作的原因。

目标是根据要求简单地删除 100 亿个实体……好吧……没那么多。我真的不想屈服于使用 mapreduce，我认为我也不需要。仍然很确定没有任务队列实现的请求会超时。因此有一个任务队列。

在我的例子中，失败的任务队列。像这样：

INFO 2018-01-28 18:48:21,129 module.py:788] default: "POST /del/text HTTP/1.1" 302 - WARNING 2018-01-28 18:48:21,129 taskqueue_stub.py:1981] Task task5 failed to execute. This task will retry in 3600.000 seconds

这是我精彩的失败代码：

@sign_in_
@app.route('/del/<entity_kind>', methods=["GET", "POST"])
def DeleteStuff(entity_kind):

  allowed_deletion = {
    'text': models.Text,
  }

  cursor = None
  bookmark = request.form.get('bookmark', '')
  if bookmark:
    cursor = ndb.Cursor.from_websafe_string(bookmark)

  query = allowed_deletion[entity_kind].query()
  entries, next_cursor, more = query.fetch_page(
    1000, 
    keys_only=True,
    start_cursor=cursor)

  ndb.delete_multi(entries)

  bookmark = None
  if more:
    bookmark = next_cursor.to_websafe_string()

  taskqueue.add(
    url='/del/'+entity_kind,
    params={'bookmark': bookmark}
    )
  return "{0} deleted".format(entity_kind)

但是，我什至不能得到一个只有 return 语句的任务来执行：

在app.yaml

- url: /del/*
  script: app.app
  login: admin

"trigger" 处理程序绑定到 app.yaml:

@sign_in_
@app.route('/del/<entity_kind>', methods=["GET", "POST"])
def DeleteStuff(entity_kind):

  allowed_deletion = {
    'text': models.Text,
    'call': models.Call,
    'voicemail': models.Voicemail,
    'image': models.Image,
    'email': models.Email,
  }


  taskqueue.add(
    url='/deleting/'+entity_kind,
    #params={'bookmark': bookmark}
    )
  return "{0} to be deleted".format(entity_kind)

worker.yaml:

runtime: python27
api_version: 1
threadsafe: true
service: worker

handlers:
- url: /deleting/.*
  script: worker.app
  login: admin

worker.py:

from flask import Flask, render_template, request, redirect, url_for, abort, make_response
from google.appengine.ext import ndb
import datetime


from app import app
from app import models

@app.route('/deleting/<entity_kind>',methods=['POST'])
def DeletingStuff(entity_kind):

  print "entered task queue"

  allowed_deletion = {
    'text': models.Text,
    'call': models.Call,
    'voicemail': models.Voicemail,
    'image': models.Image,
    'email': models.Email,
  }
  return

应用结构

app (folder)
    - __init__.py
    - app.py
    - worker.py
    - worker.yaml
    - app.yaml

也许 worker 应该在它自己的文件夹中？我不知道如何将它连接到烧瓶中...

任务队列中的打印语句甚至没有触发。

我在终端机上看到的都是

INFO 2018-01-28 22:11:21,432 module.py:788] default: "POST /deleting/image HTTP/1.1" 302 - WARNING 2018-01-28 22:11:21,432 taskqueue_stub.py:1981] Task task25 failed to execute. This task will retry in 409.600 seconds

最后，一个可行的解决方案

@sign_in_
@app.route('/del/<entity_type>', methods=["GET", "POST"])
def DeleteStuff(entity_type):

  taskqueue.add(
    url    = "/execute_task",
    method = "POST",
    params = {
      "entity_type": entity_type,
    }
  ) # from 'from google.appengine.api import taskqueue'
  return Response("sending {} task to queue... Check logs".format(entity_type), mimetype='text/plain', status=200)#Response from 'from Flask import Response'


@sign_in_
@app.route('/execute_task', methods=["GET", "POST"])
def execute_task():
  allowed_deletion = {
    'a': models.SomemodelA,
    'b': models.SomemodelB,
    'c': models.SomemodelC,
    'd': models.SomemodelD,
  }
  entity_type = request.form.get("entity_type", '')
  bookmark = request.form.get('bookmark', '')

  cursor = None
  if bookmark:
    cursor = ndb.Cursor.from_websafe_string(bookmark)

  if entity_type:
    query = allowed_deletion[entity_type].query()
    entries, next_cursor, more = query.fetch_page(
      1000, 
      keys_only=True,
      start_cursor=cursor)

    ndb.delete_multi(entries)

    bookmark = None
    if more:
      bookmark = next_cursor.to_websafe_string()

      taskqueue.add(
        url='/execute_task',
        params={'entity_type': entity_type, 'bookmark': bookmark}
      )

  return Response("did it. You'll never see this message", mimetype='text/plain', status=200)

笔记

@sign_in_ 也可以在任务中不抛出错误

Google App Engine 为任务队列抛出带有可变 url 的失败错误，换句话说，这将抛出失败错误：

@app.route('/execute_task/<some_variable>', methods=["GET", "POST"])
def execute_task(some_variable):
    pass

两个网址都在app.yaml中列出，不需要worker.py和worker.yaml

- url: /del/*
  script: app.app
  login: admin
- url: /execute_task
  script: app.app
  login: admin

非常感谢GAEfan的帮助！！ :) 另请注意：我能够从他在我的另一个 post 中给出的代码重新开始调试解决方案：

Answer 1

开始于：

@app.route('/del/<entity_kind>', methods = ['GET', 'POST'])

接下来，request.args 用于查询字符串。您要处理 POST 参数：

request.form.get('bookmark', '')

接下来，您的 @sign_in_ 标记导致任务队列重定向到登录页面。（请参阅错误日志中引用的 302？）。删除它。尝试将此添加到您的 app.yaml:

- url: /del/.*
  script: whereisapplication.app
  login: admin

这会将其置于您的 Google 登录后，但任务队列仍然可以访问。

更新你的更新：

任务中的打印语句没有意义；你永远不会看到它。尝试：

import logging
logging.info("{0} to be deleted".format(entity_kind))

或：

@app.route('/deleting/<entity_kind>/',methods=['GET', 'POST'])
@app.route('/deleting/<entity_kind>',methods=['GET', 'POST'])
def DeletingStuff(entity_kind):
    if request.method == 'GET':
        print "here"

并查看您是否可以在浏览器中访问 url。请注意第一条路线中的尾部斜杠：以帮助诊断重定向问题。

您也可以完全跳过 POST 方法，并使用查询字符串：

/deleting/text?cursor=387123481246123469

cursor = request.args.get("cursor", None)

删除实体的任务队列失败 - python - flask - google appengine

taskqueue for deleting entities fail - python - flask - google appengine

python

google-app-engine

entities

del

task-queue