Add support for asyncio

[savv]

Are there any plans to add async versions of this API?

I.e. is there/will there be a way to await on neo4j queries?

[technige]

There are no plans to add this for Python in the near future. One of the reasons is due to the complexity around supporting both Python 2.x and 3.x in the same driver, given that async functionality is Python 3 only.

[adamrothman]

@savvopoulos Right now the best solution is to execute blocking calls to Neo4j using the event loop's run_in_executor method. Something like this:

import asyncio

from neo4j.v1 import GraphDatabase


class AsyncNeo4j:

    _driver = None

    @classmethod
    def get_driver(cls):
        if cls._driver is None:
            cls._driver = GraphDatabase.driver('bolt://my-neo4j-host:7687')
        return cls._driver

    @classmethod
    async def run(cls, statement, *args, **kwargs):
        """Run a statement asynchronously using the event loop's default thread
        pool executor.
        """
        driver = cls.get_driver()

        def run():
            with driver.session() as session:
                result = session.run(statement, *args, **kwargs)
            return result

        loop = asyncio.get_event_loop()
        # Passing None uses the default executor
        return await loop.run_in_executor(None, run)

This works well enough – I use it in production – but it's obviously not ideal.

@technige It would be truly outstanding if one day the code responsible for managing the Neo4j protocol (packing queries and parameters for Bolt transport, unpacking the responses, etc.) could be separated from the transport (currently, blocking socket I/O) code.

This is precisely what Hyper-h2 does for HTTP/2. It provides a full implementation of the protocol without any I/O whatsoever. This makes it easy for third parties to wrap the protocol and supply their own I/O, whether via an asyncio.Protocol, the equivalent Twisted construct, or even some alternative synchronous transport.

I would definitely be interested in writing such a wrapper, given an appropriate protocol implementation. What do you think?

[PaulRudin]

For example, the python elasticsearch client has an adapter with a different transport for use with asycio: https://github.com/elastic/elasticsearch-py-async

[davebshow]

I've been evaluating the potential for async Python connections to Neo4j, and ended up implementing a Bolt client/server for asyncio. Pretty simple, more of a POC at this point, but I think it would be cool if community members picked up the project: https://github.com/davebshow/asyncbolt

Also, @adamrothman, the asycbolt.messaging and asyncbolt.buffer modules should be able to be used to pack and unpack Bolt messages independently of the asyncio.Protocol implementation. Any other implementation of the protocol that used them would have to implement all of the session details etc., but the structs and enums are all defined in asyncbolt.messaging, and the buffers appear to be working, though not very well tested or optimized.

[savv]

@adamrothman That's a start, but aren't you blocking while results are being fetched? I don't think the session.run(), which you run inside of run_in_executor, fetches the results. So, each time you iterate over the BoltStatementResult iterator returned by AsyncNeo4j.run(), your entire server will block while a row is being fetched.

[savv]

This worked well for me:

    async def arun(self, query):
        with self.driver.session(access_mode=neo4j.v1.READ_ACCESS) as session:
            def run():
                return session.run(query).records()
            it = await self.loop.run_in_executor(self.executor, run)

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

[adamrothman]

Ah, I didn't realize it was fetching results lazily – thanks for the tip @savvopoulos. Maybe something like this would be simpler?

@classmethod
async def run(cls, statement, *args, **kwargs):
    """Run a statement asynchronously using the event loop's default thread
    pool executor.
    """
    driver = cls.get_driver()

    def run():
        with driver.session() as session:
            result = session.run(statement, *args, **kwargs)
        return list(result.records())

    loop = asyncio.get_event_loop()
    # Passing None uses the default executor
    return await loop.run_in_executor(None, run)

Just consume all the records on the thread and return them in a list?

[savv]

For smaller datasets this is fine, otherwise you consume a lot of memory.

(By the way, still not a 100% sure if they are fetched lazily; but a few tests seem to suggest so.)

[technige]

Yes, results are fetched lazily.

[savv]

Here's the complete class, including retry logic, in case the connection is lost. I've been using this successfully for a while now.

from concurrent import futures
import neo4j
from neo4j.v1 import GraphDatabase, basic_auth, Node
import time
import traceback

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.v1.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

[technige]

I'd be happy to look at how we might introduce async to Python if there's enough interest. Let's collect 👍s on this comment...

[g3rd]

@savvopoulos Thanks for the example!

1. First off, thanks for protecting against the BrokenPipeError exception. I see that exception a lot throughout my environments.
2. I don't see where the sessions or db connection being closed. Is that by design?

Edit:
RE: 2 - Found the close()

[technige]

As the plan currently stands, we will likely remove support for Python 2 when the 2.0 series drivers are released (probably next year, in conjunction with the 4.0 server release although timescales are still very fluid).

After Python 2 had been removed from the equation, we will look at adding async support.

[technige]

Adding support for asyncio is now on our roadmap. Due to other priorites however, this will arrive at the earliest in the second half of 2019, but may not arrive until later. Note that support for Python 2 will be dropped in the first half of 2019, paving the way for this feature.

Closing this issue now that the item has been added to the roadmap.