TINKERPOP-2774: Add native-async WebSocket driver for gremlin-python

CHANGELOG.asciidoc

@@ -27,6 +27,11 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
 
 This release also includes changes from prior 3.7.x releases.
 
+* Added native-async WebSocket driver for Python (`gremlin_python.driver`): low-level transport (`AsyncAiohttpWSTransport`), protocol (`AsyncGremlinServerWSProtocol`), result streaming (`AsyncResultSet`), connection (`AsyncConnection`), and client (`AsyncClient`) — full asyncio usage without thread-pool wrappers.
+* Added `AsyncDriverRemoteConnection` for the Python GLV async driver: a high-level connection object equivalent to `DriverRemoteConnection` that exposes `submit()`, `submit_async()`, and `submit_stream()` as coroutines, supports server-side sessions, and automatically rolls back any open transaction on `close()`.
+* Added `AsyncGraphTraversalSource`, `AsyncGraphTraversal`, and `async_traversal()` factory to the Python GLV: enables the full Gremlin DSL in async mode (`await g.V().has_label("person").values("name").to_list()`); terminal steps `to_list()`, `next()`, `to_set()`, `has_next()`, and `iterate()` are all awaitable.
+* Added `AsyncTransaction` to the Python GLV: `asyncio.Lock`-protected transaction helper returned by `await g.tx()` with `begin()`, `commit()`, `rollback()`, and async context-manager support.
+* Added keepalive pings (`ping_interval`), automatic retry on connection reset (`InvalidatedConnectionError`), and per-client default request options (including `query_timeout`) to the Python GLV async driver for production reliability.
 * Enabled building and running with Java 21 and Java 25 (experimental; `spark-gremlin` excluded, as Spark 3.3.x only runs on Java 8 through 17).
 * Bumped to Groovy 4.0.32 which adds support for parsing Java 25 bytecode.
 * Bumped Hadoop to 3.4.3 (and Kerby to 2.0.3) to enable `hadoop-gremlin` to build and run on Java 25.

gremlin-python/src/main/python/examples/async_connections.py

@@ -0,0 +1,108 @@
+#
+# 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 os
+import ssl
+import sys
+import uuid
+
+sys.path.append("..")
+
+from gremlin_python.driver.async_driver_remote_connection import AsyncDriverRemoteConnection
+from gremlin_python.driver.async_graph_traversal import async_traversal
+
+SERVER_URL = os.getenv("GREMLIN_SERVER_URL", "ws://localhost:8182/gremlin").format(45940)
+VERTEX_LABEL = os.getenv("VERTEX_LABEL", "connection")
+
+
+async def main():
+    await with_context_manager()
+    await with_explicit_close()
+    await with_auth()
+    await with_custom_pool()
+    await with_session()
+
+
+async def with_context_manager():
+    # The recommended way: async context manager guarantees close() is called
+    # even if an exception is raised.
+    async with AsyncDriverRemoteConnection(SERVER_URL, "g") as rc:
+        g = async_traversal().with_remote(rc)
+
+        # Terminal steps (to_list, next, iterate) are coroutines — await them.
+        await g.add_v(VERTEX_LABEL).iterate()
+
+        count = await g.V().has_label(VERTEX_LABEL).count().next()
+        print("with_context_manager — vertex count:", count)
+
+
+async def with_explicit_close():
+    # When a context manager is not practical, close() manually in a finally block.
+    rc = AsyncDriverRemoteConnection(SERVER_URL, "g")
+    try:
+        g = async_traversal().with_remote(rc)
+        count = await g.V().count().next()
+        print("with_explicit_close — vertex count:", count)
+    finally:
+        await rc.close()
+
+
+async def with_auth():
+    # Plain-text (SASL/basic) authentication.
+    auth_url = os.getenv("GREMLIN_SERVER_BASIC_AUTH_URL",
+                         "ws://localhost:8182/gremlin").format(45941)
+
+    ssl_ctx = None
+    if ":45941" in auth_url:
+        ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ssl_ctx.check_hostname = False
+        ssl_ctx.verify_mode = ssl.CERT_NONE
+
+    async with AsyncDriverRemoteConnection(
+        auth_url, "g",
+        username="stephen", password="password",
+        ssl=ssl_ctx,
+    ) as rc:
+        g = async_traversal().with_remote(rc)
+        count = await g.V().count().next()
+        print("with_auth — vertex count:", count)
+
+
+async def with_custom_pool():
+    # pool_size controls concurrent WebSocket connections.
+    async with AsyncDriverRemoteConnection(SERVER_URL, "g", pool_size=4) as rc:
+        g = async_traversal().with_remote(rc)
+        count = await g.V().count().next()
+        print("with_custom_pool — vertex count:", count)
+
+
+async def with_session():
+    # Server-side sessions preserve state across multiple requests.
+    session_id = str(uuid.uuid4())
+    async with AsyncDriverRemoteConnection(SERVER_URL, "g", session=session_id) as rc:
+        g = async_traversal().with_remote(rc)
+
+        await g.add_v(VERTEX_LABEL).property("session", session_id).iterate()
+        count = await g.V().has("session", session_id).count().next()
+        print("with_session — vertex count:", count)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

gremlin-python/src/main/python/examples/async_traversals.py

@@ -0,0 +1,117 @@
+#
+# 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.
+#
+
+# This example requires the Modern toy graph to be preloaded when Gremlin Server
+# starts.  See https://tinkerpop.apache.org/docs/current/reference/#gremlin-server-docker-image
+# and use conf/gremlin-server-modern.yaml.
+
+import asyncio
+import os
+import sys
+
+sys.path.append("..")
+
+from gremlin_python.driver.async_driver_remote_connection import AsyncDriverRemoteConnection
+from gremlin_python.driver.async_graph_traversal import async_traversal
+from gremlin_python.process.graph_traversal import __
+from gremlin_python.process.traversal import P, T
+
+SERVER_URL = os.getenv("GREMLIN_SERVER_URL", "ws://localhost:8182/gremlin").format(45940)
+
+# CI uses port 45940 with gmodern binding; local uses 8182 with g binding.
+GRAPH_BINDING = "gmodern" if ":45940" in SERVER_URL else "g"
+
+
+async def main():
+    async with AsyncDriverRemoteConnection(SERVER_URL, GRAPH_BINDING) as rc:
+        g = async_traversal().with_remote(rc)
+
+        await basic_queries(g)
+        await write_operations(g)
+        await concurrent_queries(g)
+        await traversal_steps(g)
+
+
+async def basic_queries(g):
+    # to_list() collects all results into a list — identical name to the sync
+    # version, but here it is a coroutine that must be awaited.
+    names = await g.V().values("name").to_list()
+    print("basic_queries — all names:", names)
+
+    # next() returns the first result (raises StopAsyncIteration if empty).
+    count = await g.V().count().next()
+    print("basic_queries — vertex count:", count)
+
+    # to_set() deduplicates results.
+    labels = await g.V().label().to_set()
+    print("basic_queries — unique labels:", labels)
+
+    # has_next() returns True/False without consuming the traversal.
+    exists = await g.V().has("name", "marko").has_next()
+    print("basic_queries — marko exists:", exists)
+
+    # Filtering and projection with snake_case step names.
+    person_maps = await g.V().has_label("person").value_map("name", "age").to_list()
+    print("basic_queries — person maps:", person_maps)
+
+
+async def write_operations(g):
+    # iterate() discards results and is the right choice for writes.
+    # It adds a "discard" step so the server knows no data need be returned.
+    await g.add_v("planet").property("name", "saturn").iterate()
+
+    saturn = await g.V().has("name", "saturn").next()
+    print("write_operations — created:", saturn)
+
+    # Clean up.
+    await g.V().has("name", "saturn").drop().iterate()
+
+
+async def concurrent_queries(g):
+    # asyncio.gather() runs multiple traversals concurrently over the shared
+    # connection pool — no extra connections are opened.
+    names_coro = g.V().has_label("person").values("name").to_list()
+    edges_coro = g.E().label().to_set()
+    software_coro = g.V().has_label("software").values("name").to_list()
+
+    names, edges, software = await asyncio.gather(names_coro, edges_coro, software_coro)
+    print("concurrent_queries — people:   ", names)
+    print("concurrent_queries — edge types:", edges)
+    print("concurrent_queries — software:  ", software)
+
+
+async def traversal_steps(g):
+    # The full Gremlin DSL is available — all intermediate steps work
+    # exactly as in the sync version.
+    e1 = await g.V(1).both_e().to_list()
+    print("traversal_steps (1) — all edges of v1:", e1)
+
+    e2 = await g.V(1).both_e().where(__.other_v().has_id(2)).to_list()
+    print("traversal_steps (2) — edges between v1 and v2:", e2)
+
+    e3 = await g.V(1).out_e().where(__.in_v().has(T.id, P.within(2, 3))).to_list()
+    print("traversal_steps (3) — out-edges of v1 to v2 or v3:", e3)
+
+    # next(n) returns the first n results as a list.
+    first_two = await g.V().has_label("person").values("name").next(2)
+    print("traversal_steps — first two person names:", first_two)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

gremlin-python/src/main/python/gremlin_python/driver/aiohttp/async_transport.py

@@ -0,0 +1,145 @@
+#
+# 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 sys
+
+import aiohttp
+from aiohttp import ClientResponseError
+
+from gremlin_python.driver.transport import AbstractBaseTransport
+
+if sys.version_info >= (3, 11):
+    from asyncio import timeout as _async_timeout
+else:
+    from async_timeout import timeout as _async_timeout
+
+__author__ = 'Apache TinkerPop (dev@tinkerpop.apache.org)'
+
+
+class AsyncAiohttpWSTransport(AbstractBaseTransport):
+    """Native-async WebSocket transport built on aiohttp.
+
+    Unlike :class:`~gremlin_python.driver.aiohttp.transport.AiohttpTransport`,
+    every I/O method is a coroutine.  The transport does **not** create its own
+    event loop; callers must already be running inside one (``async def``
+    contexts, ``asyncio.run()``, etc.).
+
+    Parameters
+    ----------
+    read_timeout:
+        Maximum seconds to wait for a WebSocket frame.  ``None`` disables the
+        timeout.
+    write_timeout:
+        Maximum seconds to wait when sending a frame.  ``None`` disables the
+        timeout.
+    **kwargs:
+        Additional keyword arguments forwarded verbatim to
+        :meth:`aiohttp.ClientSession.ws_connect`.  ``max_content_length`` is
+        transparently remapped to ``max_msg_size``; ``ssl_options`` is
+        remapped to ``ssl``.
+    """
+
+    def __init__(self, read_timeout=None, write_timeout=None, **kwargs):
+        self._websocket = None
+        self._client_session = None
+        self._read_timeout = read_timeout
+        self._write_timeout = write_timeout
+        self._aiohttp_kwargs = kwargs
+        if "max_content_length" in self._aiohttp_kwargs:
+            self._aiohttp_kwargs["max_msg_size"] = self._aiohttp_kwargs.pop(
+                "max_content_length"
+            )
+        if "ssl_options" in self._aiohttp_kwargs:
+            self._aiohttp_kwargs["ssl"] = self._aiohttp_kwargs.pop("ssl_options")
+
+    async def connect(self, url, headers=None):
+        """Open the WebSocket connection to *url*.
+
+        Parameters
+        ----------
+        url:
+            WebSocket endpoint, e.g. ``ws://localhost:8182/gremlin``.
+        headers:
+            Optional extra HTTP headers sent during the WebSocket handshake.
+        """
+        self._client_session = aiohttp.ClientSession()
+        try:
+            self._websocket = await self._client_session.ws_connect(
+                url, headers=headers, **self._aiohttp_kwargs
+            )
+        except ClientResponseError as err:
+            if err.status == 403:
+                raise Exception(
+                    "Failed to connect to server: HTTP Error code 403 - Forbidden."
+                )
+            raise
+
+    async def write(self, message):
+        """Send *message* (bytes) as a WebSocket binary frame."""
+        async with _async_timeout(self._write_timeout):
+            await self._websocket.send_bytes(message)
+
+    async def read(self):
+        """Receive the next WebSocket message and return its payload as bytes.
+
+        Text frames are UTF-8 encoded before returning.  Close/error frames
+        raise :class:`RuntimeError`.
+        """
+        async with _async_timeout(self._read_timeout):
+            msg = await self._websocket.receive()
+
+        if msg.type == aiohttp.WSMsgType.CLOSE:
+            await self.close()
+            raise RuntimeError("Connection was closed by server.")
+        if msg.type == aiohttp.WSMsgType.CLOSED:
+            raise RuntimeError("Connection was already closed.")
+        if msg.type == aiohttp.WSMsgType.ERROR:
+            raise RuntimeError("Received error on read: '" + str(msg.data) + "'")
+        if msg.type == aiohttp.WSMsgType.TEXT:
+            return msg.data.strip().encode("utf-8")
+        return msg.data
+
+    async def ping(self):
+        """Send a WebSocket ping frame to keep the connection alive."""
+        await self._websocket.ping()
+
+    async def close(self):
+        """Close the WebSocket and the underlying HTTP session gracefully.
+
+        Both resources are closed concurrently and each close is shielded from
+        external cancellation so that the connection is always cleaned up even
+        if the caller's coroutine is cancelled.
+        """
+        tasks = []
+        if self._websocket is not None and not self._websocket.closed:
+            tasks.append(asyncio.ensure_future(self._websocket.close()))
+        if self._client_session is not None and not self._client_session.closed:
+            tasks.append(asyncio.ensure_future(self._client_session.close()))
+        if tasks:
+            await asyncio.gather(*[asyncio.shield(t) for t in tasks])
+
+    @property
+    def closed(self):
+        """``True`` if the WebSocket or the underlying session is closed."""
+        return (
+            self._websocket is None
+            or self._websocket.closed
+            or self._client_session is None
+            or self._client_session.closed
+        )