[
https://issues.apache.org/jira/browse/TINKERPOP-2774?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131033#comment-18131033
]
ASF GitHub Bot commented on TINKERPOP-2774:
-------------------------------------------
xiazcy commented on code in PR #3527:
URL: https://github.com/apache/tinkerpop/pull/3527#discussion_r4361882612
##########
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).
Review Comment:
This seems off to me. `discard()` isn't new in 3.8, it's the renamed
`none()` step that `iterate()` has emitted since 3.3, serving as a server-side
optimization that avoids sending large, unneeded result sets back for write
traversals. Removing it makes the server serialize and stream the full result
set just for the client to discard it, which I'd consider a regression on
`iterate()`.
There's really no compatibility need here, since the 3.8 line targets 3.8
servers and the rename is an accepted breaking change. This also deviates from
the behavior in every other driver.
So I think discard should be added back, with any relevant test changes.
> 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)
