Graphene 运行上下文管理器中的所有解析器

Question

Aiohttp 提供了一个上下文管理器来创建客户端会话。建议对多个 http 查询使用一个会话（在大多数情况下每个应用程序）https://aiohttp.readthedocs.io/en/stable/client_quickstart.html#make-a-request
但是石墨烯使用解析器需要声明为class方法：
http://docs.graphene-python.org/en/latest/execution/execute/ 对于石墨烯也存在asyncio执行器https://github.com/graphql-python/aiohttp-graphql 有什么方法可以与上下文异步执行所有解析器吗？
示例：

async def get_task(session, api_url, id):
    """ Function to resolve task from rest API"""
    async with session.get(api_url+id) as response:
        return await response.json()
class Query(graphene.ObjectType):
   task = graphene.Field(Task)
   async def resolve_task(self, info, session, id=1):
   """This method needs to be run
      in async with aiohttp.ClientSession() as session:
      context"""
        return await get_task(session, url, id)

想到了全局变量的装饰器或者中间件，但是看起来很难看。是否有更多的地产和 pythonic 方式来做到这一点？

Answer 1

我会为此使用 context。参见 https://docs.graphene-python.org/en/latest/execution/execute/

示例：

import aiohttp
import asyncio
import graphene
from graphql.execution.executors.asyncio import AsyncioExecutor
from pprint import pprint


async def get_task(session, api_url, id):
    async with session.get(api_url + str(id)) as response:
        print(f'> Retrieving {id} using session {session}')
        return await response.json()


class Query(graphene.ObjectType):

    task = graphene.Field(
        type=graphene.types.json.JSONString,
        id=graphene.Int())

    async def resolve_task(self, info, id=1):
        return await get_task(
            session=info.context['session'], 
            api_url=info.context['api_url'], 
            id=id)


schema = graphene.Schema(query=Query)


async def main():
    query = '''
        query q1 {
          t1: task(id: 1)
          t2: task(id: 2)
        }
    '''
    async with aiohttp.ClientSession() as session:
        res = await schema.execute(
            query,
            context={
                'session': session,
                'api_url': 'https://jsonplaceholder.typicode.com/todos/',
            },
            executor=AsyncioExecutor(loop=asyncio.get_running_loop()),
            return_promise=True)
        assert not res.errors, repr(res.errors)
        pprint(res.data, width=150)


if __name__ == '__main__':
    asyncio.run(main())

输出：

$ python3 example.py
> Retrieving 2 using session <aiohttp.client.ClientSession object at 0x10917bfd0>
> Retrieving 1 using session <aiohttp.client.ClientSession object at 0x10917bfd0>
OrderedDict([('t1', '{"userId": 1, "id": 1, "title": "delectus aut autem", "completed": false}'),
             ('t2', '{"userId": 1, "id": 2, "title": "quis ut nam facilis et officia qui", "completed": false}')])

Graphene 运行上下文管理器中的所有解析器

Graphene run all resolvers in a context manager

python

python-asyncio

aiohttp

contextmanager

graphene-python

Graphene 运行 上下文管理器中的所有解析器

Graphene run all resolvers in a context manager

python

python-asyncio

aiohttp

contextmanager

graphene-python

Graphene 运行上下文管理器中的所有解析器