Problème de chaînage de tâches avec curseur sur le moteur de l'application. Exception: décalage de requête trop important. Quelqu'un d'autre a ce problème?

Question

Je ne suis pas sûr si quelqu'un d'autre a ce problème, mais je reçois une exception "trop ​​grande requête offset" en utilisant un curseur pour les tâches de chaînage sur le serveur de développement appengine (pas sûr si cela se produit en direct).

L'erreur se produit lors de la demande d'un curseur après que plus de 4000 enregistrements ont été traités dans une seule requête.

Je n'étais pas au courant que les offsets avaient quelque chose à voir avec les curseurs, et c'est peut-être juste une anomalie dans sdk pour le moteur de l'application. Pour résoudre ce problème, raccourcissez le délai avant que la tâche soit différée (afin que moins d'enregistrements soient traités à la fois) ou, lorsque vous vérifiez le temps écoulé, vous pouvez également vérifier que le nombre d'enregistrements traités est toujours compris. par exemple, si time.time()> end_time ou count == 2000 .Reset count and report task. 2000 est un nombre arbitraire, je ne suis pas sûr de ce que la limite devrait être.

EDIT:

Après avoir effectué les changements mentionnés ci-dessus, l'exécution ne se termine jamais. Le code with_cursor (curseur) est appelé, mais semble commencer au début à chaque fois. Est-ce que je manque quelque chose d'évident?

Le code qui provoque l'exception est la suivante:

Le tableau "Transact" a 4800 lignes. L'erreur se produit lorsque transacts.cursor() est appelée lorsque time.time()> end_time a la valeur true. 4510 enregistrements ont été traités au moment où le curseur est demandé, ce qui semble provoquer l'erreur (sur le serveur de développement, n'ont pas été testés ailleurs).

def some_task(trans): 
    tts = db.get(trans) 
    for t in tts: 
    #logging.info('in some_task') 
    pass 


def test_cursor(request): 
    ret = test_cursor_task() 


def test_cursor_task(cursor = None): 
    startDate = datetime.datetime(2010,7,30) 
    endDate = datetime.datetime(2010,8,30) 
    end_time = time.time() + 20.0 
    transacts = Transact.all().filter('transactionDate >', startDate).filter('transactionDate <=',endDate) 

    count =0 
    if cursor: 
     transacts.with_cursor(cursor) 
    trans =[] 
    logging.info('queue_trans') 
    for tran in transacts: 
    count+=1 
    #trans.append(str(tran)) 
    trans.append(str(tran.key())) 

    if len(trans)==20: 
      deferred.defer(some_task, trans, _countdown = 500)     
      trans =[]    
    if time.time() > end_time: 
     logging.info(count)    
     if len(trans)>0:     
      deferred.defer(some_task, trans, _countdown = 500) 
      trans =[] 
     logging.info('time limit exceeded setting next call to queue') 
     cursor = transacts.cursor() 
     deferred.defer(test_cursor_task, cursor) 
     logging.info('returning false') 
     return False 


    return True 


    return HttpResponse('') 

Espérons que cela aide quelqu'un.

Merci Bert

Answer 1

Essayez cette fois sans utiliser le iter fonctionnalité:

#... 
CHUNK = 500 
objs = transacts.fetch(CHUNK) 

for tran in objs: 
    do_your_stuff 

if len(objs) == CHUNK: 
    deferred.defer(my_task_again, cursor=str(transacts.cursor())) 

Cela fonctionne pour moi.