Flask异步路由处理程序能否适配gql的AIOHTTPTransport异步调用？

报错核心原因 #

默认Flask的WSGI运行模式（即app.run()启动的方式）对异步路由的处理机制会导致该问题：默认模式下，每个异步路由请求会被分配到独立的线程，每个线程会创建单独的asyncio事件循环，你在main函数外层提前创建的gql session绑定的是外层启动时的主事件循环，和请求处理时的事件循环不属于同一个，才会触发跨循环的Future报错。

解决方案1：使用ASGI服务器运行Flask（官方推荐） #

将Flask应用作为ASGI应用启动，保证整个应用的所有异步逻辑都运行在同一个全局事件循环中，即可正常复用提前创建的gql session，同时还能复用底层HTTP连接池，性能最优。

1. 安装依赖：

pip install uvicorn

2. 调整后代码：

from flask import Flask
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
import uvicorn

token = 'bearer ...'  # GitHub API token

# 全局初始化gql客户端，只执行一次schema拉取
transport = AIOHTTPTransport(
    url='https://api.github.com/graphql',
    headers={'Authorization': token}
)
client = Client(transport=transport, fetch_schema_from_transport=True)

app = Flask(__name__)

@app.route('/')
async def home():
    query = gql('''
    query example ($username: String!) {
        user(login: $username){
            bio,
        }
    }
    ''')
    params = {
        'username': 'octocat'
    }
    # 直接复用全局客户端执行请求，无需重建
    result = await client.execute(query, variable_values=params)
    return result

if __name__ == '__main__':
    # 用uvicorn启动ASGI应用，全局共用同一个事件循环
    uvicorn.run(app, host="127.0.0.1", port=5000)

解决方案2：预缓存Schema，兼容原有WSGI模式 #

如果必须保留默认app.run()的WSGI运行模式，可以提前预拉取Schema缓存，后续新建客户端时直接使用本地缓存的Schema，不会触发重复拉取，性能损耗可忽略：

import asyncio
from flask import Flask
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport

token = 'bearer ...'  # GitHub API token

transport = AIOHTTPTransport(
    url='https://api.github.com/graphql',
    headers={'Authorization': token}
)
# 启动时预拉取一次Schema，全局缓存
cached_schema = asyncio.run(
    Client(transport=transport, fetch_schema_from_transport=True).load_schema()
)

app = Flask(__name__)

@app.route('/')
async def home():
    query = gql('''
    query example ($username: String!) {
        user(login: $username){
            bio,
        }
    }
    ''')
    params = {
        'username': 'octocat'
    }
    # 新建客户端直接传入缓存的Schema，不会重复拉取
    client = Client(transport=transport, schema=cached_schema)
    async with client as session:
        result = await session.execute(query, variable_values=params)
        return result

if __name__ == '__main__':
    app.run()

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