首页 文章

石墨烯在上下文管理器中运行所有解析器

提问于
浏览 次
1

Aiohttp提供了一个上下文管理器来创建客户端会话 . 建议每个http查询使用一个会话（在大多数情况下每个应用程序）https://aiohttp.readthedocs.io/en/stable/client_quickstart.html#make-a-request
但石墨烯使用旋转变压器需要声明为类方法：
http://docs.graphene-python.org/en/latest/execution/execute/对于石墨烯也存在asyncio Actuator 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方式来做到这一点？

python python-asyncio aiohttp contextmanager graphene-python

1 回答

  • 0

    我会用 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}')])
    回复于

相关问题