Graphene run all resolvers in a context manager

Question

Aiohttp provides a context manager to create client session. It's recommended to use one session per many http queries ( in most cases per application ) https://aiohttp.readthedocs.io/en/stable/client_quickstart.html#make-a-request
But graphene uses resolvers needs to be declared as a class method:
http://docs.graphene-python.org/en/latest/execution/execute/ For graphene also exists asyncio executor https://github.com/graphql-python/aiohttp-graphql Is there any way to execute all resolvers in async with context?
Example:

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)

I think about decorator or middleware with global variable, but it looks ugly. Is there more estate and pythonic way to do it?

Answer 1

I would use context for that. See https://docs.graphene-python.org/en/latest/execution/execute/

Example:

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())

Output:

$ 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}')])