[
https://issues.apache.org/jira/browse/TINKERPOP-2774?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131225#comment-18131225
]
ASF GitHub Bot commented on TINKERPOP-2774:
-------------------------------------------
masterhugo commented on code in PR #3527:
URL: https://github.com/apache/tinkerpop/pull/3527#discussion_r4365243727
##########
gremlin-python/src/main/python/gremlin_python/driver/async_graph_traversal.py:
##########
@@ -0,0 +1,392 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""Async traversal layer for gremlin-python.
+
+Drop-in async replacement for the standard ``traversal()`` factory::
+
+ async with AsyncDriverRemoteConnection("ws://localhost:8182/gremlin", "g")
as rc:
+ g = async_traversal().with_remote(rc)
+
+ names = await g.V().has_label("person").values("name").to_list()
+ count = await g.V().count().next()
+ await g.add_v("person").property("name", "saturn").iterate()
+
+ async with g.tx() as gtx:
+ await gtx.add_v("person").property("name", "pluto").iterate()
+ # commits on clean exit, rolls back on exception
+"""
+import asyncio
+import logging
+
+from gremlin_python.process.graph_traversal import GraphTraversal,
GraphTraversalSource
+from gremlin_python.process.traversal import (
+ Bytecode, Traversal, TraversalStrategies, TraversalStrategy,
+)
+from gremlin_python.structure.graph import Graph
+
+log = logging.getLogger("gremlinpython")
+
+__author__ = 'Apache TinkerPop ([email protected])'
+
+
+# ---------------------------------------------------------------------------
+# Public factory
+# ---------------------------------------------------------------------------
+
+def async_traversal():
+ """Return an :class:`_AsyncAnonymousTraversalSource`.
+
+ Use in place of
:func:`~gremlin_python.process.anonymous_traversal.traversal`
+ when working with
:class:`~gremlin_python.driver.async_driver_remote_connection.AsyncDriverRemoteConnection`.
+ """
+ return _AsyncAnonymousTraversalSource()
+
+
+class _AsyncAnonymousTraversalSource:
+ def with_remote(self, remote_connection):
+ """Bind *remote_connection* and return an
:class:`AsyncGraphTraversalSource`."""
+ return AsyncGraphTraversalSource(
+ Graph(),
+ AsyncTraversalStrategies(),
+ remote_connection=remote_connection,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Strategy pipeline (async)
+# ---------------------------------------------------------------------------
+
+class AsyncTraversalStrategies(TraversalStrategies):
+ """Async-aware strategy pipeline.
+
+ Calls ``await strategy.apply(traversal)`` for coroutine strategies and
+ ``strategy.apply(traversal)`` for plain sync strategies so both can
coexist.
+ """
+
+ async def apply_strategies(self, traversal): # noqa: D102
+ for strategy in self.traversal_strategies:
+ if asyncio.iscoroutinefunction(strategy.apply):
+ await strategy.apply(traversal)
+ else:
+ strategy.apply(traversal)
+
+
+class AsyncRemoteStrategy(TraversalStrategy):
+ """Client-side strategy that submits bytecode to the remote server
asynchronously.
+
+ Registered with the ``py:AsyncRemoteStrategy`` fqcn so it is never sent
+ to the server as part of the serialized traversal.
+ """
+
+ def __init__(self, remote_connection):
+ super().__init__(fqcn="py:AsyncRemoteStrategy")
+ self.remote_connection = remote_connection
+
+ async def apply(self, traversal):
+ if traversal.traversers is None:
+ remote_traversal = await
self.remote_connection.submit(traversal.bytecode)
+ traversal.remote_results = remote_traversal
+ traversal.traversers = remote_traversal.traversers
+
+
+# ---------------------------------------------------------------------------
+# Traversal base (async terminal steps)
+# ---------------------------------------------------------------------------
+
+class AsyncTraversal(Traversal):
+ """Base traversal class that replaces blocking terminal steps with
coroutines.
+
+ Concrete subclasses must be used with :class:`AsyncTraversalStrategies` so
+ that ``apply_strategies`` can be awaited inside ``__anext__``.
+ """
+
+ # Prevent accidental sync iteration.
+ def __next__(self):
+ raise NotImplementedError(
+ "AsyncTraversal does not support synchronous iteration. "
+ "Use 'async for', 'await to_list()', or 'await next()'."
+ )
+
+ # ------------------------------------------------------------------
+ # Async iteration protocol
+ # ------------------------------------------------------------------
+
+ def __aiter__(self):
+ return self
+
+ async def __anext__(self):
+ if self.traversers is None:
+ await self.traversal_strategies.apply_strategies(self)
+ if self.last_traverser is None:
+ # StopAsyncIteration propagates naturally when the result set is
exhausted.
+ self.last_traverser = await self.traversers.__anext__()
+ obj = self.last_traverser.object
+ self.last_traverser.bulk -= 1
+ if self.last_traverser.bulk <= 0:
+ self.last_traverser = None
+ return obj
+
+ # ------------------------------------------------------------------
+ # Async terminal steps
+ # ------------------------------------------------------------------
+
+ async def to_list(self):
+ """Execute the traversal and return all results as a :class:`list`."""
+ results = []
+ async for item in self:
+ results.append(item)
+ return results
+
+ async def to_set(self):
+ """Execute the traversal and return all results as a :class:`set`."""
+ return set(await self.to_list())
+
+ async def next(self, amount=None):
+ """Execute and return the first result, or the first *amount* results.
+
+ Raises :class:`StopAsyncIteration` when *amount* is ``None`` and the
+ result set is empty.
+ """
+ if amount is None:
+ return await self.__anext__()
+ results = []
+ for _ in range(amount):
+ try:
+ results.append(await self.__anext__())
+ except StopAsyncIteration:
+ break
+ return results
+
+ async def iterate(self):
+ """Execute the traversal, discarding all results.
+
+ Use for write operations (``add_v``, ``add_e``, ``property``, …)
+ where no return value is needed.
+
+ Results are drained on the client side for compatibility with servers
+ that predate the ``discard()`` step (e.g. Neptune, TinkerPop < 3.8.0).
+ """
+ async for _ in self:
+ pass
+ return self
+
+ async def has_next(self):
+ """Return ``True`` if the traversal produces at least one result."""
+ try:
+ await self.__anext__()
Review Comment:
Good catch, fixed! Updated has_next() and next_traverser() to match the sync
Traversal behavior and added tests for it.
> Full async/await support in gremlin-python
> ------------------------------------------
>
> Key: TINKERPOP-2774
> URL: https://issues.apache.org/jira/browse/TINKERPOP-2774
> Project: TinkerPop
> Issue Type: Improvement
> Components: python
> Affects Versions: 3.5.3
> Reporter: Ken Hu
> Priority: Minor
>
> Recently, there have been multiple instances of users asking about issues
> with event loops and aiohttp. This discussion nearly always ends up with the
> user requesting for async/await support in gremlin-python. This means full
> compatibility with asyncio and not needing to use threads.
> Async/await support is also a feature that those using newer, asynchronous
> web frameworks are asking for as gremlin-python can be a performance
> bottleneck due to its lack of support for async/await.
> My intention is that we can use this issue as a means of gauging the interest
> from the community for this feature. If you are interested in having
> gremlin-python support async/await then please leave a comment below.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
