Le "yield" du générateur Python dans une fonction distincte

Question

J'implémente une bibliothèque utilitaire qui est une sorte de gestionnaire de tâches destiné à fonctionner dans l'environnement distribué du service de cloud computing Google App Engine. (Il utilise une combinaison de files d'attente de tâches et de memcache pour exécuter le traitement de fond). Je prévois d'utiliser des générateurs pour contrôler l'exécution des tâches, en imposant essentiellement une "concurrence" non-préemptive via l'utilisation de yield dans le code de l'utilisateur.

L'exemple trivial - le traitement d'un groupe d'entités de base de données - pourrait être quelque chose comme ce qui suit :

class EntityWorker(Worker):
    def setup():
        self.entity_query = Entity.all()
    def run():
        for e in self.entity_query:
            do_something_with(e)
            yield

Comme nous le savons, yield est un canal de communication bidirectionnel, permettant de transmettre des valeurs au code qui utilise les générateurs. Ceci permet de simuler une "API préemptive" telle que la SLEEP appel ci-dessous :

def run():
    for e in self.entity_query:
        do_something_with(e)
        yield Worker.SLEEP, timedelta(seconds=1)

Mais c'est moche. Ce serait bien de cacher le yield dans une fonction distincte qui peut être invoquée de manière simple :

self.sleep(timedelta(seconds=1))

Le problème est que mettre yield en fonction sleep tourne il en une fonction de générateur. L'appel ci-dessus renverrait donc simplement un autre générateur. Ce n'est qu'après avoir ajouté .next() y yield en arrière, nous obtiendrions le résultat précédent :

yield self.sleep(timedelta(seconds=1)).next()

qui est bien sûr encore plus laid et inutilement verbeux qu'avant.

D'où ma question : Y a-t-il un moyen de mettre yield en fonction sans la transformer en fonction de générateur mais en la rendant utilisable par d'autres générateurs pour produire des valeurs calculées par elle ?

Answer 1

Vous semblez ne pas comprendre l'évidence :

class EntityWorker(Worker):
    def setup(self):
        self.entity_query = Entity.all()

    def run(self):
        for e in self.entity_query:
            do_something_with(e)
            yield self.sleep(timedelta(seconds=1))

    def sleep(self, wait):
        return Worker.SLEEP, wait

C'est le yield qui transforme les fonctions en générateurs, il est impossible de le laisser de côté.

Pour cacher le rendement, vous avez besoin d'une fonction d'ordre supérieur, dans votre exemple c'est map :

from itertools import imap

def slowmap(f, sleep, *iters):
    for row in imap(f, self.entity_query):
        yield Worker.SLEEP, wait

def run():
    return slowmap(do_something_with, 
                   (Worker.SLEEP, timedelta(seconds=1)),
                   self.entity_query)

Answer 2

Hélas, ça ne marchera pas. Mais un "juste milieu" pourrait convenir :

def sleepjob(*a, **k):
    if a:
        return Worker.SLEEP, a[0]
    else:
        return Worker.SLEEP, timedelta(**k)

Alors

yield self.sleepjob(timedelta(seconds=1))
yield self.sleepjob(seconds=1)

ça me semble correct.

Answer 3

Je vous suggère de jeter un coup d'œil à la ndb . Il utilise les générateurs comme co-routines (comme vous le proposez ici), ce qui vous permet d'écrire des programmes qui fonctionnent avec les rpcs de manière asynchrone.

Pour ce faire, l'API enveloppe le générateur avec une autre fonction qui "amorce" le générateur (elle appelle immédiatement .next() pour que le code commence à être exécuté). Les tasklets sont également conçus pour fonctionner avec l'infrastructure rpc d'App Engine, ce qui permet d'utiliser n'importe lequel des appels d'api asynchrones existants.

Avec le modèle de concurrencie utilisé dans ndb, on peut yield soit un objet futur (similaire à ce qui est décrit dans pep-3148) ou un objet rpc de l'App Engine. Lorsque ce rpc est terminé, l'exécution dans la fonction qui a produit l'objet est autorisée à continuer.

Si vous utilisez un modèle dérivé de ndb.model.Model alors ce qui suit vous permettra d'itérer de manière asynchrone sur une requête :

from ndb import tasklets

@tasklets.tasklet
def run():
it = iter(Entity.query())
# Other tasklets will be allowed to run if the next call has to wait for an rpc.
while (yield it.has_next_async()):
  entity = it.next()
  do_something_with(entity)

Bien que ndb soit encore considéré comme expérimental (une partie de son code de gestion des erreurs nécessite encore du travail), je vous recommande d'y jeter un œil. Je l'ai utilisé dans mes deux derniers projets et j'ai trouvé que c'était une excellente bibliothèque.

Assurez-vous de lire la documentation liée à la page principale, ainsi que la documentation d'accompagnement de l'outil de gestion de l'environnement. trucs de tasklet .