python之如何覆盖异步 NDB 方法并编写自己的 tasklet

我正在尝试掌握 NDB 引入的异步操作，我想使用 @ndb.tasklet 来异步我的一些工作。

简单的例子是在重写的 get_or_insert_async 中生成 string_id

这是处理事情的正确方法吗？这里有什么可以改进的？

@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) 

另一个例子是以扇出的方式在循环中做一些事情。这是正确的吗？

@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) 

编辑:

按照整个概念的理解，yield 在这里提供一个 Future 对象，然后并行收集(如果可能)并异步执行。我说得对吗？

在 Nick 的提示之后(是您的意思吗？):

@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 

编辑:

新修订版……

@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 

请您参考如下方法:

如果你想做的只是用不同的参数调用一些异步的东西，你不需要使用 tasklets - 只需返回包装函数的返回值，就像这样:

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

不过，出于以下几个原因，我对此持谨慎态度:您正在更改内置函数的含义，这通常是个坏主意，您正在更改签名(位置参数但没有关键字参数)，并且您没有将额外的参数传递给原始函数。

对于你的第二个例子，一次产生一个东西将迫使 NDB 等待它们完成——'yield' 是 'wait' 的同义词。相反，为集合中的每个元素执行 tasklet 函数，然后同时等待所有元素(通过调用列表上的 yield)。