Python asyncio协程二次返回GraphQL查询结果时触发RuntimeError

问题 #

我编写了一个GraphQLLibrary类，核心功能是创建GraphQL客户端、异步发送GraphQL请求（查询/变更）并返回结果。后续需要支持订阅和触发订阅的变更，因此请求必须异步执行。

首次创建客户端并发送查询一切正常，但第二次发送查询时，虽然在send_graphql_request_coroutine内已获取到结果，但返回时触发RuntimeError，错误信息：

RuntimeError: <asyncio.locks.Event object at 0x00000191E01F55D0 [unset]> is bound to a different event loop

我原以为第一次调用asyncio.run(...)后事件循环会关闭，无法理解该错误原因。

代码示例 #

from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport
import asyncio


class GraphQLLibrary:
    def __init__(self):
        self.client = None

    def send_graphql_request(self, query): 
        query_result = asyncio.run(self.send_graphql_request_coroutine(query))
        print("===== received query result: ", query_result)
        return query_result

    def create_graphql_client(self, url):
        transport = WebsocketsTransport(url)
        self.client = Client(transport=transport)

    async def send_graphql_request_coroutine(self, query):
        if not self.client: 
            raise AssertionError("GraphQL client is not initialized. Call 'Create GraphQL Client' first.")
        
        async with self.client as session:      
            query = gql(query)

            query_result = await session.execute(query)          
            print("===== got query result:", query_result, "\n===== returning ...")
            return query_result


url = "ws://URL/"

query = """query { system { version }}"""

gql_lib = GraphQLLibrary()

gql_lib.create_graphql_client(url)
gql_lib.send_graphql_request(query)     # 第一次请求正常
gql_lib.send_graphql_request(query)     # 第二次请求返回结果时出错

终端错误输出 #

===== got query result: {'system': {'version': 'SR355.SC42.D.24090913'}} 
===== returning ...
===== received query result:  {'system': {'version': 'SR355.SC42.D.24090913'}}
===== got query result: {'system': {'version': 'SR355.SC42.D.24090913'}} 
===== returning ...
Traceback (most recent call last):
  File "c:\Users\User\directory\test_gqlLib.py", line 55, in <module>
    gql_lib.send_graphql_request(query)     # this raises an error, when returning the query result
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\Users\User\directory\test_gqlLib.py", line 19, in send_graphql_request
    query_result = asyncio.run(self.send_graphql_request_coroutine(query))
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\base_events.py", line 650, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "c:\Users\User\directory\test_gqlLib.py", line 34, in send_graphql_request_coroutine
    async with self.client as session:
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\User\directory\.venv\Lib\site-packages\gql\client.py", line 662, in __aexit__
    await self.close_async()
  File "C:\Users\User\directory\.venv\Lib\site-packages\gql\client.py", line 654, in close_async
    await self.transport.close()
  File "C:\Users\User\directory\.venv\Lib\site-packages\gql\transport\websockets_base.py", line 660, in close
    await self.wait_closed()
  File "C:\Users\User\directory\.venv\Lib\site-packages\gql\transport\websockets_base.py", line 667, in wait_closed
    await self._wait_closed.wait()
  File "C:\Program Files\Python311\Lib\asyncio\locks.py", line 210, in wait
    fut = self._get_loop().create_future()
          ^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\mixins.py", line 20, in _get_loop
    raise RuntimeError(f'{self!r} is bound to a different event loop')
RuntimeError: <asyncio.locks.Event object at 0x00000191E01F55D0 [unset]> is bound to a different event loop

原因分析 #

问题根源在于asyncio.run()的特性与WebsocketsTransport的内部实现冲突：

• 每次调用asyncio.run()都会创建一个全新的事件循环，执行完毕后自动关闭该循环。
• 第一次请求时，WebsocketsTransport内部的_wait_closed事件对象绑定了第一次的事件循环。
• 第二次调用asyncio.run()会创建新的事件循环，但客户端复用了之前的WebsocketsTransport实例，当请求结束关闭客户端时，尝试在新循环中等待绑定旧循环的_wait_closed事件，导致循环不匹配的错误。

尽管第一次请求后事件循环已关闭，但WebsocketsTransport实例并未被销毁，其内部事件对象仍保留对旧循环的引用。

解决方案 #

方案1：复用同一个事件循环 #

修改GraphQLLibrary类，初始化时创建并保留一个持久化的事件循环，所有请求都在该循环中执行：

from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport
import asyncio


class GraphQLLibrary:
    def __init__(self):
        self.client = None
        self.loop = asyncio.new_event_loop()  # 创建持久化事件循环

    def send_graphql_request(self, query): 
        query_result = self.loop.run_until_complete(self.send_graphql_request_coroutine(query))
        print("===== received query result: ", query_result)
        return query_result

    def create_graphql_client(self, url):
        transport = WebsocketsTransport(url)
        self.client = Client(transport=transport)

    async def send_graphql_request_coroutine(self, query):
        if not self.client: 
            raise AssertionError("GraphQL client is not initialized. Call 'Create GraphQL Client' first.")
        
        async with self.client as session:      
            query = gql(query)
            query_result = await session.execute(query)          
            print("===== got query result:", query_result, "\n===== returning ...")
            return query_result

    def close(self):
        # 程序结束时手动关闭事件循环
        self.loop.close()


# 使用示例
url = "ws://URL/"
query = """query { system { version }}"""

gql_lib = GraphQLLibrary()
gql_lib.create_graphql_client(url)
gql_lib.send_graphql_request(query)
gql_lib.send_graphql_request(query)
gql_lib.close()  # 最后关闭循环

方案2：每次请求重新创建客户端 #

如果不需要长期保持连接，可每次发送请求前重新创建客户端，避免复用绑定旧循环的Transport实例：

def send_graphql_request(self, query, url): 
    # 每次请求重新创建客户端
    self.create_graphql_client(url)
    query_result = asyncio.run(self.send_graphql_request_coroutine(query))
    print("===== received query result: ", query_result)
    return query_result

# 使用示例
gql_lib.send_graphql_request(query, url)
gql_lib.send_graphql_request(query, url)

方案3：全异步风格改造（推荐，适配后续订阅功能） #

既然后续要支持订阅，采用全异步写法更合理，避免同步代码嵌套异步调用：

from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport
import asyncio


class GraphQLLibrary:
    def __init__(self):
        self.client = None

    async def create_graphql_client(self, url):
        transport = WebsocketsTransport(url)
        self.client = Client(transport=transport)

    async def send_graphql_request(self, query): 
        if not self.client: 
            raise AssertionError("GraphQL client is not initialized. Call 'create_graphql_client' first.")
        
        async with self.client as session:      
            query = gql(query)
            query_result = await session.execute(query)          
            print("===== got query result:", query_result, "\n===== returning ...")
            return query_result


# 使用示例
async def main():
    url = "ws://URL/"
    query = """query { system { version }}"""

    gql_lib = GraphQLLibrary()
    await gql_lib.create_graphql_client(url)
    result1 = await gql_lib.send_graphql_request(query)
    print("===== received query result: ", result1)
    result2 = await gql_lib.send_graphql_request(query)
    print("===== received query result: ", result2)

asyncio.run(main())

内容的提问来源于stack exchange，提问作者Peid Pideijsen