Взаимодействие ndb.toplevel и ndb.transactional

Question

Можно ли получить @ndb.toplevel, чтобы играть красиво с @ndb.transactional?

То, что я хотел бы достичь, - это транзакция, которая содержит entity.put_async() звонки, но с удобством не нужно явно ждать фьючерсов. @ndb.toplevel обычно делает это, но другой вопрос SO, похоже, не может быть объединен с транзакциями: "Does ndb toplevel break transactions?"

Я не могу найти это явно документированное где-либо в документах App Engine. Мы можем воспроизвести ошибку утверждения, указанную в этом вопросе, но мы написали несколько тестов, чтобы проверить, не вызвали ли вызовы put_async() и не видели никаких проблем. Однако, поскольку есть возможность потерять данные, было бы неплохо получить более конкретный ответ от того, кто хорошо знает ndb.

Наш простой тестовый код приведен ниже. Если мы удалим как ndb.toplevel, так и декоратор ndb.transactional, тест завершится неудачно, как и ожидалось. Однако, если мы просто используем декоратор ndb.transactional и оставьте вне декоратора ndb.toplevel, проходит тест, который не ожидается. Это заставляет меня беспокоиться о том, что, возможно, в достаточно накладных расходов, что вызов put_async() имеет достаточно времени для завершения, но без какого-либо gaurantee, поэтому он может непредсказуемо провалиться?

class AsyncTestModel(ndb.Model): 
    data = ndb.StringProperty(indexed=True) 

@ndb.toplevel 
def start_test(): 
    for _ in range(100): 
     test() 

    # Check we wrote all the entities 
    time.sleep(30) 
    entities = AsyncTestModel.query().fetch() 
    assert(len(entities) == 1000) 


@ndb.transactional(xg=True) 
def test(): 
    for _ in range(10): 
     x = AsyncTestModel() 
     x.data = make_random_string(1000) 
     x.put_async() 

Answer 1

Тот факт, что тест пройден, когда вы используете только @ndb.transactional ожидаемое какой тренинг: для того, чтобы сделки никогда не применять частично, @ndb.transactional ждать, что все запросы выполняются.

Таким образом, код for _ in range(100): test() ожидает завершения транзакции на каждой итерации.

Таким образом, следующий тест пройден:

@ndb.transactional(xg=True) 
def test(): 
    for _ in range(10): 
     x = AsyncTestModel() 
     x.data = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(1000)) 
     x.put_async() 

for x, _ in enumerate(range(100)): 
    test() 
    assert(AsyncTestModel.query().count() == (x + 1) * 10) 

Примечание: если вы используете асинхронные запросы в транзакции, вы можете посмотреть на @ndb.transactional_async (документация here).