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

Question

我不知道我在做什么。

这就是我的 google appengine 任务队列不工作的原因。

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

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

INFO 2018-01-28 18:48:21,129 module.py:788] 默认值：“POST /del/text HTTP/1.1”302 - 警告 2018-01-28 18:48:21,129 taskqueue_stub.py:1981] 任务 task5 未能执行。此任务将在 3600.000 秒后重试

这是我精彩的失败代码：

@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

“触发器”处理程序绑定到 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

也许工人应该在它自己的文件夹中？不过我不知道怎么把它挂在烧瓶里……

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

我在终端中得到的只是一些内容

INFO 2018-01-28 22:11:21,432 module.py:788] 默认值：“POST /deleting/image HTTP/1.1" 302 - 警告 2018-01-28 22:11:21,432 taskqueue_stub.py:1981] 任务 task25 未能执行。该任务将 在 409.600 秒后重试

终于找到了一个可行的解决方案

@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 应用引擎会抛出任务队列的可变 url 失败错误，换句话说，这将抛出失败错误：

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

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

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

非常感谢GAEfan的帮助！！ :) 另请注意：我能够从他在我的另一篇文章中给出的代码开始调试解决方案：simple google app engine taskqueue tutoral? flask / python / GAE

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

这将把它放在你的谷歌登录后面，但任务队列仍然可以访问。

更新您的更新：

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

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)