Wie man eine asynchrone NDB-Methode überschreibt und ein eigenes Tasklet schreibt

Question

Ich versuche asynchrone Operationen zu verstehen, die mit NDB eingeführt wurden, ich möchte @ndb.tasklet verwenden, um einige meiner Arbeiten zu synchronisieren.

Das einfache Beispiel string_id Generation in den überschriebenen get_or_insert_async

würde Ist das ein richtiger Weg, um die Dinge? Was kann hier verbessert werden?

@classmethod 
@ndb.tasklet 
def get_or_insert_async(cls, *args): 
    id = cls.make_string_id(*args) 
    model = yield super(MyModel, cls).get_or_insert_async(id) 
    raise ndb.Return(model) 

Ein anderes Beispiel wäre, Dinge in einer Schleife in Fan-out irgendwie zu tun. Ist das richtig?

@classmethod 
@ndb.tasklet 
def do_stuff(cls, some_collection): 

    @ndb.tasklet 
    def internal_tasklet(data): 
     do_some_long_taking_stuff(data) 
     id = make_stuff_needed_for_id(data) 
     model = yield cls.get_or_insert_async(id) 
     model.long_processing(data) 
     yield model.put_async() 
     raise ndb.Return(None) 

    for data in some_collection: 
     # will it parallelise internal_tasklet execution? 
     yield internal_tasklet(data) 

    raise ndb.Return(None) 

EDIT:

Wie das gesamte Konzept zu verstehen, sind yields hier ein Future Objekte zu schaffen, die dann parallel gesammelt werden (wenn möglich) und asynchron ausgeführt. Hab ich recht?

Nach Nicks Hinweis (ist es, was Sie gemeint?):

@classmethod 
@ndb.tasklet 
def do_stuff(cls, some_collection): 

    @ndb.tasklet 
    def internal_tasklet(data): 
     do_some_long_taking_stuff(data) 
     id = make_stuff_needed_for_id(data) 
     model = yield cls.get_or_insert_async(id) 
     model.long_processing(data) 
     raise ndb.Return(model)    # change here 

    models = [] 
    for data in some_collection: 
     # will it parallelise internal_tasklet execution? 
     m = yield internal_tasklet(data)  # change here 
     models.appedn(m)      # change here 

    keys = yield ndb.put_multi_async(models) # change here 
    raise ndb.Return(keys)      # change here 

EDIT:

Neue überarbeitete Version ...

@classmethod 
@ndb.tasklet 
def do_stuff(cls, some_collection): 

    @ndb.tasklet 
    def internal_tasklet(data): 
     do_some_long_taking_stuff(data) 
     id = make_stuff_needed_for_id(data) 
     model = yield cls.get_or_insert_async(id) 
     model.long_processing(data) 
     raise ndb.Return(model)     

    futures = [] 
    for data in some_collection: 
     # tasklets won't run in parallel but while 
     # one is waiting on a yield (and RPC underneath) 
     # the other will advance it's execution 
     # up to a next yield or return 
     fut = internal_tasklet(data))   # change here 
     futures.append(fut)     # change here 

    Future.wait_all(futures)     # change here 

    models = [fut.get_result() for fut in futures] 
    keys = yield ndb.put_multi_async(models) # change here 
    raise ndb.Return(keys)      # change here 

Answer 1

Sie brauchen nicht Tasklets, wenn alle zu verwenden, Sie möchten etwas asynchron mit verschiedenen Argumenten aufrufen - geben Sie einfach den Rückgabewert der umgebrochenen Funktion zurück:

def get_or_insert_async(cls, *args): 
    id = cls.make_string_id(*args) 
    return super(MyModel, cls).get_or_insert_async(id) 

Ich würde jedoch aus mehreren Gründen vorsichtig sein: Sie ändern die Bedeutung einer eingebauten Funktion, was normalerweise eine schlechte Idee ist, Sie ändern die Signatur (Positionsargumente, aber keine Schlüsselwortargumente) , und Sie übergeben keine zusätzlichen Argumente an die ursprüngliche Funktion.

Für Ihr zweites Beispiel, erzwingen die Dinge einzeln nacheinander NDB auf ihre Fertigstellung warten - "Ausbeute" ist gleichbedeutend mit "warten". Führen Sie stattdessen die Tasklet-Funktion für jedes Element in der Sammlung aus, und warten Sie gleichzeitig auf alle (durch Aufrufen von yield in der Liste).