From 3470b43bc53f57d6a62f4597b5842366f931658e Mon Sep 17 00:00:00 2001 From: Mark Larah Date: Tue, 28 Jul 2026 12:23:32 -0500 Subject: [PATCH 1/2] Add xfail test for concurrent use of a shared Client instance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Demonstrates that AIOHTTPTransport.connect() raises TransportAlreadyConnected when two async-with blocks enter the same Client concurrently — a common pattern when the Client is cached (e.g. via lru_cache) for connection pooling. Co-Authored-By: Claude --- tests/test_aiohttp.py | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/tests/test_aiohttp.py b/tests/test_aiohttp.py index 00bd8a0f..d0c7e172 100644 --- a/tests/test_aiohttp.py +++ b/tests/test_aiohttp.py @@ -1,3 +1,4 @@ +import asyncio import io import json import os @@ -1918,3 +1919,41 @@ async def handler(request): await session.execute("qmlsdkfj") assert "request should be a GraphQLRequest object" in str(exc_info.value) + + +@pytest.mark.asyncio +@pytest.mark.xfail( + reason="transport.connect() is not reentrant — shared client fails on concurrent use" +) +async def test_aiohttp_reentrant_connect(aiohttp_server): + """A single Client/transport instance should support concurrent sessions. + + When a Client is cached (e.g. via lru_cache) and used from multiple + concurrent async-with blocks, the second caller should not get + TransportAlreadyConnected. + """ + from aiohttp import web + + from gql.transport.aiohttp import AIOHTTPTransport + + async def handler(request): + return web.Response(text=query1_server_answer, content_type="application/json") + + app = web.Application() + app.router.add_route("POST", "/", handler) + server = await aiohttp_server(app) + + url = server.make_url("/") + + transport = AIOHTTPTransport(url=url, timeout=10) + client = Client(transport=transport) + + results = [] + + async def use_client(): + async with client as session: + query = gql(query1_str) + result = await session.execute(query) + results.append(result) + + await asyncio.gather(use_client(), use_client()) From b3a66e6d69556370836cf519e0e1915de9b3bd35 Mon Sep 17 00:00:00 2001 From: Mark Larah Date: Tue, 28 Jul 2026 12:25:07 -0500 Subject: [PATCH 2/2] Make AIOHTTPTransport.connect() reentrant via refcount A cached Client instance (common when wrapping with lru_cache for connection pooling) would raise TransportAlreadyConnected if entered concurrently from multiple async-with blocks, or if a prior session wasn't cleaned up due to an abrupt disconnection. connect() now reuses the existing aiohttp.ClientSession when one is already active (incrementing a reference count), and close() only tears it down when the last caller exits. This matches aiohttp's own recommendation of a single long-lived session per application. Co-Authored-By: Claude --- gql/transport/aiohttp.py | 16 ++++++++++++++-- tests/test_aiohttp.py | 14 +++++++------- 2 files changed, 21 insertions(+), 9 deletions(-) diff --git a/gql/transport/aiohttp.py b/gql/transport/aiohttp.py index e7eff55f..f7f48577 100644 --- a/gql/transport/aiohttp.py +++ b/gql/transport/aiohttp.py @@ -98,6 +98,7 @@ def __init__( self.ssl_close_timeout: Optional[Union[int, float]] = ssl_close_timeout self.client_session_args = client_session_args self.session: Optional[aiohttp.ClientSession] = None + self._connect_count: int = 0 self.response_headers: Optional[CIMultiDictProxy[str]] self.json_serialize: Callable = json_serialize self.json_deserialize: Callable = json_deserialize @@ -110,6 +111,10 @@ async def connect(self) -> None: to create the session. Should be cleaned with a call to the close coroutine. + + This method is reentrant: if the session already exists, the existing + session is reused and a reference count is incremented. Each connect() + call must be paired with a close() call. """ if self.session is None: @@ -136,8 +141,7 @@ async def connect(self) -> None: self.session = aiohttp.ClientSession(**client_session_args) - else: - raise TransportAlreadyConnected("Transport is already connected") + self._connect_count += 1 async def close(self) -> None: """Coroutine which will close the aiohttp session. @@ -145,7 +149,15 @@ async def close(self) -> None: Don't call this coroutine directly on the transport, instead use :code:`async with` on the client and this coroutine will be executed when you exit the async context manager. + + The session is only actually closed when every connect() call has been + balanced by a close() call. """ + self._connect_count = max(0, self._connect_count - 1) + + if self._connect_count > 0: + return + if self.session is not None: log.debug("Closing transport") diff --git a/tests/test_aiohttp.py b/tests/test_aiohttp.py index d0c7e172..f80d9865 100644 --- a/tests/test_aiohttp.py +++ b/tests/test_aiohttp.py @@ -392,7 +392,7 @@ async def handler(request): @pytest.mark.asyncio -async def test_aiohttp_cannot_connect_twice(aiohttp_server): +async def test_aiohttp_connect_twice_is_reentrant(aiohttp_server): from aiohttp import web from gql.transport.aiohttp import AIOHTTPTransport @@ -409,9 +409,12 @@ async def handler(request): transport = AIOHTTPTransport(url=url, timeout=10) async with Client(transport=transport) as session: - - with pytest.raises(TransportAlreadyConnected): - await session.transport.connect() + # Second connect should succeed (reentrant) and reuse the session + original_session = session.transport.session + await session.transport.connect() + assert session.transport.session is original_session + # Balance the extra connect with a close + await session.transport.close() @pytest.mark.asyncio @@ -1922,9 +1925,6 @@ async def handler(request): @pytest.mark.asyncio -@pytest.mark.xfail( - reason="transport.connect() is not reentrant — shared client fails on concurrent use" -) async def test_aiohttp_reentrant_connect(aiohttp_server): """A single Client/transport instance should support concurrent sessions.