Asyncio python - TypeError: требуется будущее, сопрограмма или ожидание - PullRequest
Новая
       10

Asyncio python - TypeError: требуется будущее, сопрограмма или ожидание

0 голосов
/ 08 января 2020

Одна из функций asyn c возвращает объект генератора asyn c. Я добавил l oop .run_until_complete (fun c ()), но, тем не менее, он выдает ошибку как «TypeError: требуется будущее, сопрограмма или ожидание». Ниже приведен код. Я пытаюсь получить записи из Neo4j асинхронно. Я получил асин c "Neo4j класс от GitHub. Я новичок в этой асин c концепции. 

from concurrent import futures
import neo4j
from neo4j import GraphDatabase, basic_auth
import time
import traceback
import asyncio 

RETRY_WAITS = [0, 1, 4]  # How long to wait after each successive failure.
class Neo4j:
    """Neo4j database API."""
    def __init__(self, config, loop):
        self.config = config        
        self.loop = loop
        self.executor = futures.ThreadPoolExecutor(max_workers=30)
        for retry_wait in RETRY_WAITS:
            try:
                self.init_driver()
                break
            except:
                if retry_wait == RETRY_WAITS[-1]:
                    raise
                else:
                    print('WARNING: retrying to Init DB; err:')
                    traceback.print_exc()
                    time.sleep(retry_wait)  # wait for 0, 1, 3... seconds.

    def init_driver(self):
        auth = basic_auth(self.config['user'], self.config['pass'])
        self.driver = GraphDatabase.driver(self.config['url'], auth=auth)

    async def afetch_start(self, query):
        session = self.driver.session(access_mode=neo4j.READ_ACCESS)
        def run():
            return session.run(query).records()
        return session, await self.loop.run_in_executor(self.executor, run)

    async def afetch_iterate(self, session, iter):
            def iterate():
                try:
                    return next(iter)
                except StopIteration:
                    return None
            while True:
                res = await self.loop.run_in_executor(self.executor, iterate)
                if res is None:
                    return
                else:
                    yield dict(res)

    async def afetch(self, query):
        for retry_wait in RETRY_WAITS:
            try:
                session, iter = await self.afetch_start(query)
                break
            except (BrokenPipeError, neo4j.exceptions.ServiceUnavailable) as e:
                if retry_wait == RETRY_WAITS[-1]:
                    raise
                else:
                    await asyncio.sleep(retry_wait)
                    await self.loop.run_in_executor(self.executor, self.init_driver)
        async for x in self.afetch_iterate(session, iter):
            yield x

        await self.loop.run_in_executor(self.executor, session.close)


    async def afetch_one(self, query):
        async for i in self.afetch(query):
            return i
        return None

    async def aexec(self, query):
        async for i in self.afetch(query):
            pass
        return


config={'url':"bolt://localhost",'user':'neo4j','pass':'pwd'}    
loop=asyncio.get_event_loop()    
n=Neo4j(config,loop)    
loop.run_until_complete(n.afetch("MATCH(p:Person)-[:Acted_in]->(mv:Movies) RETURN p.name as actors"))
loop.close()

- РЕДАКТИРОВАТЬ

Я изменил код для правильной работы. Запрос возвращает 218 тыс. Строк, и для извлечения полного списка требуется 5 минут, а та же операция asyn c в C# завершается всего за 2 се c. Похоже на приведенный выше код до сих пор не go в асин c

1 Ответ

0 голосов
/ 08 января 2020

Очень трудно сказать, что именно происходит без воспроизводимого примера , но я сделаю предположение. Вы, вероятно, передаете asyn c объект генератора в al oop, вы не должны этого делать. Способ работы с генераторами asyn c заключается в использовании async for. Вот пример: 

import asyncio


async def func():  # async generator
    yield 1
    yield 2
    yield 3


async def main():
    async for i in func():  # get values from async generator
        print(i)


asyncio.run(main())  # can be used instead of loop.run_until_complete(main())
...