Graphene запускает все преобразователи в диспетчере контекста

Aiohttp предоставляет менеджер контекста для создания сеанса клиента. Рекомендуется использовать один сеанс на множество http-запросов (в большинстве случаев для каждого приложения) https://aiohttp.readthedocs.io/en/stable/client_quickstart.html#make-a-request
Но графен использует распознаватели, которые необходимо объявить как метод класса:
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)

Я думаю о декораторе или промежуточном программном обеспечении с глобальной переменной, но это выглядит некрасиво. Есть ли больше недвижимости и питонического способа сделать это?


python graphene-python python-asyncio aiohttp contextmanager
person stepuncius    schedule 26.02.2018    source источник

Ответы (1)


arrow_upward
3
arrow_downward

Я бы использовал 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}')])
person Messa    schedule 13.01.2019