[
https://issues.apache.org/jira/browse/TINKERPOP-2774?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130699#comment-18130699
]
ASF GitHub Bot commented on TINKERPOP-2774:
-------------------------------------------
kenhuuu commented on code in PR #3527:
URL: https://github.com/apache/tinkerpop/pull/3527#discussion_r3876534901
##########
gremlin-python/src/main/python/gremlin_python/driver/async_client.py:
##########
@@ -0,0 +1,290 @@
+#
+# 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.
+#
+import asyncio
+import logging
+
+from gremlin_python.driver import protocol, request, serializer
+from gremlin_python.driver.async_connection import AsyncConnection,
InvalidatedConnectionError
+from gremlin_python.driver.async_protocol import AsyncGremlinServerWSProtocol
+from gremlin_python.driver.aiohttp.async_transport import
AsyncAiohttpWSTransport
+from gremlin_python.driver.client import Client
+from gremlin_python.process import traversal
+
+log = logging.getLogger("gremlinpython")
+
+__author__ = 'Apache TinkerPop ([email protected])'
+
+
+class AsyncClient(Client):
+ """Native-async Gremlin client using WebSocket transport.
+
+ Unlike :class:`~gremlin_python.driver.client.Client`, no
Review Comment:
Nit: I'm guessing these are Sphinx style comments. We don't use that
regularly for gremlin-python so they look out of place. It would be best to
keep the style the same through gremlin-python. If it can't be easily changed
then its fine to leave it the same since its harmless.
##########
gremlin-python/src/main/python/gremlin_python/driver/async_resultset.py:
##########
@@ -0,0 +1,148 @@
+#
+# 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.
+#
+import asyncio
+import queue
+
+from gremlin_python.driver.resultset import ResultSet
+
+__author__ = 'Apache TinkerPop ([email protected])'
+
+# Internal sentinel used to signal that the stream is exhausted.
+_EXHAUSTED = object()
+
+
+class AsyncResultSet(ResultSet):
+ """Result set backed by an :class:`asyncio.Queue` and an
+ :class:`asyncio.Future`.
+
+ Consumption methods (:meth:`one`, :meth:`all`) are coroutines. The class
+ also supports ``async for`` iteration.
+
+ The ``future`` attribute **must** be set to a resolved
+ :class:`asyncio.Future` or a running :class:`asyncio.Task` by the owning
+ :class:`~gremlin_python.driver.async_connection.AsyncConnection` before
+ any consumption method is called.
+ """
+
+ def __init__(self, stream, request_id):
+ # stream must be an asyncio.Queue
+ super().__init__(stream, request_id)
+ self._future = None
+ # Cache individual items already unpacked from a multi-item chunk so
+ # that successive one() calls each return exactly one result.
+ self._item_cache = queue.Queue()
+
+ # ------------------------------------------------------------------
+ # future property (asyncio.Future / asyncio.Task)
+ # ------------------------------------------------------------------
+
+ @property
+ def future(self):
+ return self._future
+
+ @future.setter
+ def future(self, value):
+ self._future = value
+
+ # ------------------------------------------------------------------
+ # Disable the synchronous interface
+ # ------------------------------------------------------------------
+
+ def __iter__(self):
+ raise NotImplementedError(
+ "AsyncResultSet supports only async iteration; use 'async for'"
+ )
+
+ def __next__(self):
+ raise NotImplementedError(
+ "AsyncResultSet supports only async iteration; use 'async for'"
+ )
+
+ # ------------------------------------------------------------------
+ # Async iteration protocol
+ # ------------------------------------------------------------------
+
+ def __aiter__(self):
+ return self
+
+ async def __anext__(self):
+ result = await self.one()
+ if result is _EXHAUSTED:
+ raise StopAsyncIteration
+ return result
+
+ # ------------------------------------------------------------------
+ # Consumption
+ # ------------------------------------------------------------------
+
+ async def one(self):
+ """Return the next individual result or the ``_EXHAUSTED`` sentinel.
+
+ Each server response chunk contains a list of items. ``one()``
+ unpacks those lists so that each call returns exactly one item,
Review Comment:
This is technically a behavioral difference from the sync implementation. I
actually think this makes sense but it prevents us from directly saying its a
drop-in replacement.
##########
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:
Wouldn't this cause `has_next` to accidentally consume a result?
##########
gremlin-python/src/main/python/gremlin_python/driver/async_driver_remote_connection.py:
##########
@@ -0,0 +1,297 @@
+#
+# 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.
+#
+import asyncio
+import logging
+import uuid
+
+from gremlin_python.driver.async_client import AsyncClient
+from gremlin_python.driver.remote_connection import RemoteConnection
+from gremlin_python.process.strategies import OptionsStrategy
+from gremlin_python.process.traversal import Bytecode
+
+log = logging.getLogger("gremlinpython")
+
+__author__ = 'Apache TinkerPop ([email protected])'
+
+
+class AsyncDriverRemoteConnection(RemoteConnection):
+ """Async remote connection backed by
:class:`~gremlin_python.driver.async_client.AsyncClient`.
+
+ Primary entry point for the async Gremlin driver. Works with
+ :func:`~gremlin_python.driver.async_graph_traversal.async_traversal` to
+ provide a fully async traversal DSL::
+
+ 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()
+
+ **Robustness features**
+
+ * Per-connection WebSocket keepalive pings (see *ping_interval*).
+ * Automatic retry on network resets (handled inside
+ :class:`~gremlin_python.driver.async_client.AsyncClient`).
+ * *query_timeout* sets a default ``evaluationTimeout`` on every request.
+ * Session connections always attempt a rollback on ``close()`` to prevent
+ dangling transactions if the caller forgets to commit or roll back.
+
+ Parameters
+ ----------
+ url:
+ WebSocket endpoint, e.g. ``ws://localhost:8182/gremlin``.
+ traversal_source:
+ Gremlin traversal source name (default ``"g"``).
+ protocol_factory:
+ Forwarded to :class:`~gremlin_python.driver.async_client.AsyncClient`.
+ transport_factory:
+ Forwarded to :class:`~gremlin_python.driver.async_client.AsyncClient`.
+ pool_size:
+ Number of concurrent WebSocket connections (default 8).
+ message_serializer:
+ Serializer instance (default GraphBinarySerializersV1).
+ username, password:
+ Credentials for SASL/basic authentication.
+ kerberized_service:
+ Kerberos service principal.
+ headers:
+ Extra HTTP headers for every WebSocket handshake.
+ session:
+ Session UUID string to enable server-side sessions.
+ enable_user_agent_on_connect:
+ Send the driver user-agent header (default ``True``).
+ ping_interval:
+ Seconds between keepalive pings (default 60, 0 to disable).
+ query_timeout:
+ Default query timeout in **seconds** applied to every request as
+ ``evaluationTimeout``. Per-request options override this.
+ **transport_kwargs:
+ Extra keyword arguments forwarded to the default transport.
+ """
+
+ def __init__(
Review Comment:
This seems to be missing `enable_compression`, `graphson_reader`, and
`graphson_writer` from the `DriverRemoteConnection`. Any particular reason for
that? Compression is the big one since it can heavily affect performance (even
with GraphBinary).
##########
gremlin-python/src/main/python/gremlin_python/driver/async_client.py:
##########
@@ -0,0 +1,290 @@
+#
+# 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.
+#
+import asyncio
+import logging
+
+from gremlin_python.driver import protocol, request, serializer
+from gremlin_python.driver.async_connection import AsyncConnection,
InvalidatedConnectionError
+from gremlin_python.driver.async_protocol import AsyncGremlinServerWSProtocol
+from gremlin_python.driver.aiohttp.async_transport import
AsyncAiohttpWSTransport
+from gremlin_python.driver.client import Client
+from gremlin_python.process import traversal
+
+log = logging.getLogger("gremlinpython")
+
+__author__ = 'Apache TinkerPop ([email protected])'
+
+
+class AsyncClient(Client):
+ """Native-async Gremlin client using WebSocket transport.
+
+ Unlike :class:`~gremlin_python.driver.client.Client`, no
+ :class:`~concurrent.futures.ThreadPoolExecutor` is created. The
+ connection pool is an :class:`asyncio.Queue` and all submit/close
+ methods are coroutines that must be awaited.
+
+ Robustness features:
+
+ * Each connection runs a background WebSocket ping to prevent idle
+ disconnects (configurable via ``ping_interval``).
+ * On
:class:`~gremlin_python.driver.async_connection.InvalidatedConnectionError`
+ the request is automatically retried on a fresh connection up to
+ *pool_size* times.
+ * *default_request_options* are merged into every request so options
+ like ``evaluationTimeout`` can be set once per client.
+
+ Usage::
+
+ async with AsyncClient("ws://localhost:8182/gremlin", "g") as client:
+ result_set = await client.submit("g.V().count()")
+ print(await result_set.all())
+
+ Parameters
+ ----------
+ url:
+ WebSocket endpoint, e.g. ``ws://localhost:8182/gremlin``.
+ traversal_source:
+ Gremlin traversal source name (default ``"g"``).
+ protocol_factory:
+ Zero-argument callable returning an
+
:class:`~gremlin_python.driver.async_protocol.AsyncGremlinServerWSProtocol`.
+ transport_factory:
+ Zero-argument callable returning an
+
:class:`~gremlin_python.driver.aiohttp.async_transport.AsyncAiohttpWSTransport`.
+ pool_size:
+ Number of concurrent WebSocket connections (default 8; forced to 1
+ in session mode).
+ message_serializer:
+ Serializer to use (default
+ :class:`~gremlin_python.driver.serializer.GraphBinarySerializersV1`).
+ username, password:
+ Credentials for SASL/basic authentication.
+ kerberized_service:
+ Kerberos service principal for Kerberos authentication.
+ headers:
+ Extra HTTP headers included in every WebSocket handshake.
+ session:
+ Session UUID string to enable server-side sessions.
+ enable_user_agent_on_connect:
+ Send the driver user-agent header (default ``True``).
+ ping_interval:
+ Seconds between WebSocket keepalive pings (default 60, 0 to disable).
+ default_request_options:
+ Dict of request options merged into every request
+ (e.g. ``{"evaluationTimeout": 30000}``). Per-request options take
+ precedence over defaults.
+ **transport_kwargs:
+ Extra keyword arguments forwarded to the default
+
:class:`~gremlin_python.driver.aiohttp.async_transport.AsyncAiohttpWSTransport`.
+ """
+
+ def __init__(
+ self,
+ url,
+ traversal_source,
+ protocol_factory=None,
+ transport_factory=None,
+ pool_size=None,
+ max_workers=None, # accepted but unused — no thread pool
+ message_serializer=None,
+ username="",
+ password="",
+ kerberized_service="",
+ headers=None,
+ session=None,
+ enable_user_agent_on_connect=True,
+ ping_interval=60,
+ default_request_options=None,
+ **transport_kwargs,
+ ):
+ log.info("Creating AsyncClient with url '%s'", url)
+
+ # Do NOT call super().__init__() — Client.__init__ creates a
+ # ThreadPoolExecutor and a threading queue, neither of which we want.
+ self._closed = False
+ self._url = url
+ self._headers = headers
+ self._enable_user_agent_on_connect = enable_user_agent_on_connect
+ self._traversal_source = traversal_source
+ self._executor = None # no thread pool; attribute expected by Client
+ self._ping_interval = ping_interval
+ self._default_request_options = default_request_options or {}
+
+ if "max_content_length" not in transport_kwargs:
+ transport_kwargs["max_content_length"] = 10 * 1024 * 1024
+
+ if message_serializer is None:
+ message_serializer = serializer.GraphBinarySerializersV1()
+
+ self._message_serializer = message_serializer
+ self._username = username
+ self._password = password
+ self._session = session
+ self._session_enabled = session is not None and session != ""
+
+ if transport_factory is None:
+ def transport_factory():
+ return AsyncAiohttpWSTransport(**transport_kwargs)
+
+ self._transport_factory = transport_factory
+
+ if protocol_factory is None:
+ def protocol_factory():
+ return AsyncGremlinServerWSProtocol(
+ self._message_serializer,
+ username=self._username,
+ password=self._password,
+ kerberized_service=kerberized_service,
+ max_content_length=transport_kwargs["max_content_length"],
+ )
+
+ self._protocol_factory = protocol_factory
+
+ if self._session_enabled:
+ if pool_size is None:
+ pool_size = 1
+ elif pool_size != 1:
+ raise Exception("PoolSize must be 1 on session mode!")
+ if pool_size is None:
+ pool_size = 8
+ self._pool_size = pool_size
+
+ self._pool = asyncio.Queue()
+ self._fill_pool()
+
+ # ------------------------------------------------------------------
+ # Internal helpers
+ # ------------------------------------------------------------------
+
+ def _get_connection(self):
+ proto = self._protocol_factory()
+ return AsyncConnection(
+ self._url,
+ self._traversal_source,
+ proto,
+ self._transport_factory,
+ None,
+ self._pool,
+ headers=self._headers,
+ enable_user_agent_on_connect=self._enable_user_agent_on_connect,
+ ping_interval=self._ping_interval,
+ )
+
+ # ------------------------------------------------------------------
+ # Lifecycle
+ # ------------------------------------------------------------------
+
+ async def close(self):
+ """Close all pooled connections and mark the client as closed."""
+ if self._closed:
+ return
+ if self._session_enabled:
+ await self._close_session()
+ log.info("Closing AsyncClient with url '%s'", self._url)
+ conns = []
+ while not self._pool.empty():
+ conns.append(self._pool.get_nowait())
+ await asyncio.gather(*(conn.close() for conn in conns),
return_exceptions=True)
+ self._closed = True
+
+ async def _close_session(self):
+ message = request.RequestMessage(
+ processor="session",
+ op="close",
+ args={"session": str(self._session)},
+ )
+ conn = await self._pool.get()
+ try:
+ result_set = await conn.write(message)
+ await result_set.all()
+ except protocol.GremlinServerError:
+ pass
+
+ # ------------------------------------------------------------------
+ # Context manager
+ # ------------------------------------------------------------------
+
+ async def __aenter__(self):
+ return self
+
+ async def __aexit__(self, *_):
+ await self.close()
+
+ # ------------------------------------------------------------------
+ # Submit (with retry)
+ # ------------------------------------------------------------------
+
+ async def submit(self, message, bindings=None, request_options=None):
+ """Submit *message* and return an
:class:`~gremlin_python.driver.async_resultset.AsyncResultSet`.
+
+ Retries automatically up to *pool_size* times when a connection is
+ invalidated mid-request due to a network reset.
+ """
+ last_error = None
+ for _ in range(self._pool_size):
Review Comment:
The GLVs intentionally stay away from attempting retries as its not safe to
do so. The recommendation is that users have to add retry into their
applications. If you really want this then you will need to keep this behind a
connection option and keep it disabled by default so its opt in.
##########
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__()
+ return True
+ except StopAsyncIteration:
+ return False
+
+ async def next_traverser(self):
+ """Return the next raw
:class:`~gremlin_python.process.traversal.Traverser`."""
+ if self.traversers is None:
+ await self.traversal_strategies.apply_strategies(self)
+ return await self.traversers.__anext__()
+
+
+# ---------------------------------------------------------------------------
+# Concrete traversal classes
+# ---------------------------------------------------------------------------
+
+class AsyncGraphTraversal(GraphTraversal, AsyncTraversal):
+ """Async graph traversal.
+
+ Inherits all Gremlin DSL steps from
:class:`~gremlin_python.process.graph_traversal.GraphTraversal`
+ and async terminal steps from :class:`AsyncTraversal` via Python's MRO.
+
+ MRO: AsyncGraphTraversal → GraphTraversal → AsyncTraversal → Traversal →
object
+
+ Because ``GraphTraversal`` does not define terminal steps directly (it
+ inherits them from ``Traversal``), Python finds ``AsyncTraversal``\'s
+ async versions first in the MRO before reaching ``Traversal``\'s sync
+ versions.
+ """
+
+
+class AsyncRemoteTraversal(AsyncTraversal):
+ """Holds the :class:`~gremlin_python.driver.async_resultset.AsyncResultSet`
+ returned by
:meth:`~gremlin_python.driver.async_driver_remote_connection.AsyncDriverRemoteConnection.submit`.
+
+ ``traversers`` is set directly to the result set; strategies are never
+ applied (there is nothing to apply on an already-executed traversal).
+ """
+
+ def __init__(self, traversers):
+ super().__init__(None, None, None)
+ self.traversers = traversers
+
+
+# ---------------------------------------------------------------------------
+# Traversal source
+# ---------------------------------------------------------------------------
+
+class AsyncGraphTraversalSource(GraphTraversalSource):
+ """Async traversal source.
+
+ All spawned traversals are :class:`AsyncGraphTraversal` instances.
+ Configuration steps (``with_bulk()``, ``with_path()``, …) return a new
+ :class:`AsyncGraphTraversalSource`.
+
+ Use :func:`async_traversal` to create one::
+
+ g = async_traversal().with_remote(rc)
+ """
+
+ def __init__(self, graph, traversal_strategies, bytecode=None,
remote_connection=None):
+ # Call super WITHOUT remote_connection to avoid RemoteStrategy being
added.
+ super().__init__(graph, traversal_strategies, bytecode)
+ self.graph_traversal = AsyncGraphTraversal
+ self.remote_connection = remote_connection
+ if remote_connection is not None:
+
self.traversal_strategies.add_strategies([AsyncRemoteStrategy(remote_connection)])
+
+ def get_graph_traversal_source(self):
+ """Return a copy of this source (used by configuration steps)."""
+ # Strategies are copied — do NOT pass remote_connection to avoid
+ # registering AsyncRemoteStrategy a second time.
+ return AsyncGraphTraversalSource(
+ self.graph,
+ AsyncTraversalStrategies(self.traversal_strategies),
+ Bytecode(self.bytecode),
+ )
+
+ def get_graph_traversal(self):
+ """Return a fresh :class:`AsyncGraphTraversal` for this source."""
+ return AsyncGraphTraversal(
+ self.graph,
+ self.traversal_strategies,
+ Bytecode(self.bytecode),
+ )
+
+ async def tx(self):
+ """Return an :class:`AsyncTransaction` for this connection.
+
+ Raises if the connection is already session-bound.
+ """
+ strategy = next(
+ (s for s in self.traversal_strategies.traversal_strategies
+ if isinstance(s, AsyncRemoteStrategy)),
+ None,
+ )
+ if strategy is None:
+ raise Exception("No remote connection configured on this traversal
source")
+ if strategy.remote_connection.is_session_bound():
+ raise Exception(
+ "Cannot open a transaction on an already session-bound
connection"
+ )
+ return AsyncTransaction(self, strategy.remote_connection)
+
+
+# ---------------------------------------------------------------------------
+# Transaction
+# ---------------------------------------------------------------------------
+
+class AsyncTransaction:
+ """Async transaction over a session-bound connection.
+
+ Supports the async context-manager protocol::
+
+ async with await g.tx() as gtx:
+ await gtx.add_v("person").property("name", "pluto").iterate()
+ # commits on clean exit; rolls back on exception
+
+ Or manage it explicitly::
+
+ tx = await g.tx()
+ gtx = await tx.begin()
+ try:
+ await gtx.add_v("person").property("name", "pluto").iterate()
+ await tx.commit()
+ except Exception:
+ await tx.rollback()
+ """
+
+ def __init__(self, g, remote_connection):
+ self._g = g
+ self._remote_connection = remote_connection
+ self._session_connection = None
+ self._gtx = None
+ self._is_open = False
+ self._mutex = asyncio.Lock()
+
+ @property
+ def is_open(self):
+ return self._is_open
Review Comment:
This checks only local state which could be stale?
##########
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__()
+ return True
+ except StopAsyncIteration:
+ return False
+
+ async def next_traverser(self):
+ """Return the next raw
:class:`~gremlin_python.process.traversal.Traverser`."""
+ if self.traversers is None:
+ await self.traversal_strategies.apply_strategies(self)
+ return await self.traversers.__anext__()
+
+
+# ---------------------------------------------------------------------------
+# Concrete traversal classes
+# ---------------------------------------------------------------------------
+
+class AsyncGraphTraversal(GraphTraversal, AsyncTraversal):
+ """Async graph traversal.
+
+ Inherits all Gremlin DSL steps from
:class:`~gremlin_python.process.graph_traversal.GraphTraversal`
+ and async terminal steps from :class:`AsyncTraversal` via Python's MRO.
+
+ MRO: AsyncGraphTraversal → GraphTraversal → AsyncTraversal → Traversal →
object
+
+ Because ``GraphTraversal`` does not define terminal steps directly (it
+ inherits them from ``Traversal``), Python finds ``AsyncTraversal``\'s
+ async versions first in the MRO before reaching ``Traversal``\'s sync
+ versions.
+ """
+
+
+class AsyncRemoteTraversal(AsyncTraversal):
+ """Holds the :class:`~gremlin_python.driver.async_resultset.AsyncResultSet`
+ returned by
:meth:`~gremlin_python.driver.async_driver_remote_connection.AsyncDriverRemoteConnection.submit`.
+
+ ``traversers`` is set directly to the result set; strategies are never
+ applied (there is nothing to apply on an already-executed traversal).
+ """
+
+ def __init__(self, traversers):
+ super().__init__(None, None, None)
+ self.traversers = traversers
+
+
+# ---------------------------------------------------------------------------
+# Traversal source
+# ---------------------------------------------------------------------------
+
+class AsyncGraphTraversalSource(GraphTraversalSource):
+ """Async traversal source.
+
+ All spawned traversals are :class:`AsyncGraphTraversal` instances.
+ Configuration steps (``with_bulk()``, ``with_path()``, …) return a new
+ :class:`AsyncGraphTraversalSource`.
+
+ Use :func:`async_traversal` to create one::
+
+ g = async_traversal().with_remote(rc)
+ """
+
+ def __init__(self, graph, traversal_strategies, bytecode=None,
remote_connection=None):
+ # Call super WITHOUT remote_connection to avoid RemoteStrategy being
added.
+ super().__init__(graph, traversal_strategies, bytecode)
+ self.graph_traversal = AsyncGraphTraversal
+ self.remote_connection = remote_connection
+ if remote_connection is not None:
+
self.traversal_strategies.add_strategies([AsyncRemoteStrategy(remote_connection)])
+
+ def get_graph_traversal_source(self):
+ """Return a copy of this source (used by configuration steps)."""
+ # Strategies are copied — do NOT pass remote_connection to avoid
+ # registering AsyncRemoteStrategy a second time.
+ return AsyncGraphTraversalSource(
+ self.graph,
+ AsyncTraversalStrategies(self.traversal_strategies),
+ Bytecode(self.bytecode),
+ )
+
+ def get_graph_traversal(self):
+ """Return a fresh :class:`AsyncGraphTraversal` for this source."""
+ return AsyncGraphTraversal(
+ self.graph,
+ self.traversal_strategies,
+ Bytecode(self.bytecode),
+ )
+
+ async def tx(self):
+ """Return an :class:`AsyncTransaction` for this connection.
+
+ Raises if the connection is already session-bound.
+ """
+ strategy = next(
+ (s for s in self.traversal_strategies.traversal_strategies
+ if isinstance(s, AsyncRemoteStrategy)),
+ None,
+ )
+ if strategy is None:
+ raise Exception("No remote connection configured on this traversal
source")
+ if strategy.remote_connection.is_session_bound():
+ raise Exception(
+ "Cannot open a transaction on an already session-bound
connection"
+ )
+ return AsyncTransaction(self, strategy.remote_connection)
+
+
+# ---------------------------------------------------------------------------
+# Transaction
+# ---------------------------------------------------------------------------
+
+class AsyncTransaction:
+ """Async transaction over a session-bound connection.
+
+ Supports the async context-manager protocol::
+
+ async with await g.tx() as gtx:
+ await gtx.add_v("person").property("name", "pluto").iterate()
+ # commits on clean exit; rolls back on exception
+
+ Or manage it explicitly::
+
+ tx = await g.tx()
+ gtx = await tx.begin()
+ try:
+ await gtx.add_v("person").property("name", "pluto").iterate()
+ await tx.commit()
+ except Exception:
+ await tx.rollback()
+ """
+
+ def __init__(self, g, remote_connection):
+ self._g = g
+ self._remote_connection = remote_connection
+ self._session_connection = None
+ self._gtx = None
+ self._is_open = False
+ self._mutex = asyncio.Lock()
+
+ @property
+ def is_open(self):
+ return self._is_open
+
+ @property
+ def gtx(self):
+ if self._gtx is None:
+ raise Exception("Transaction not started — call begin() first")
+ return self._gtx
+
+ async def begin(self):
+ """Open a session and return a session-bound
:class:`AsyncGraphTraversalSource`."""
+ async with self._mutex:
+ if self._is_open:
+ raise Exception("Transaction is already open")
+ self._session_connection = self._remote_connection.create_session()
+ strategies = AsyncTraversalStrategies()
+
strategies.add_strategies([AsyncRemoteStrategy(self._session_connection)])
+ self._gtx = AsyncGraphTraversalSource(
+ self._g.graph,
+ strategies,
+ )
+ self._is_open = True
+ return self._gtx
+
+ async def commit(self):
+ """Commit the transaction and close the session."""
+ async with self._mutex:
+ if not self._is_open:
+ raise Exception("Transaction is not open")
+ await self._session_connection.commit()
+ await self._close_session()
Review Comment:
I think `self._close_session()` is going to call `rollback()` so there's a
bit of an odd sequence here where a commit will also attempt rollback.
> 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)
