using async resolvers for third-party api calls

[Schayik]

Hi, I'm trying to do some api calls outside of django with graphene-django. However, I cannot figure out how to do this with the GraphQLView from graphene_django.views in urls.py.

schema.py

import graphene
import requests

url = 'https://www.passwordrandom.com/query?command=int&format=json'

class RandomNumberType(graphene.ObjectType):
    number = graphene.Int()

    async def resolve_number(self, info):
        session = info.context["session"]
        async with session.get(url) as response:
            result = await response.json()
            return result['random'][0]

class Query(graphene.ObjectType):
    random_numbers = graphene.List(RandomNumberType)

    def resolve_random_numbers(self, info):
        return [RandomNumberType()] * 5

schema = graphene.Schema(query=Query)

This schema works when I run it in a simple test by executing python async_resolvers_test.py. It will print {'randomNumbers': [{'number': 70}, {'number': 2}, {'number': 34}, {'number': 34}, {'number': 70}]} within a second as the individual requests are called parallel.

async_resolvers_test.py:

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

from async_resolvers.schema import schema

async def main():
    query = """
        {
          randomNumbers {
              number
          }
        }
    """
    async with aiohttp.ClientSession() as session:
        res = await schema.execute(
            query,
            context={"session": session,},
            executor=AsyncioExecutor(loop=asyncio.get_running_loop()),
            return_promise=True,
        )
        assert not res.errors, repr(res.errors)
        print(res.data)

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

Now, I would like to do the same thing for every execution with graphene-django, but I cannot work out how to do this. I have tried many things, but nothing works (for example adding executor=AsyncExecutor() to GraphQLView.as_view(...)).

urls.py:

from django.contrib import admin
from django.urls import path
from graphene_django.views import GraphQLView

urlpatterns = [
    path('admin/', admin.site.urls),
    path("graphql", GraphQLView.as_view(graphiql=True)),
]

I am getting this error: There is no current event loop in thread 'Thread-1', when I try to execute in GraphiQL:

{
  randomNumbers {
    number
  }
}

I am sure this has something to do with not properly setting up an async session, but I get very confused with everything that is out there and I thus have no idea how to do this.

Please let me know if you have any idea or if anything here is unclear. Thanks in advance!

[lovetoburnswhen]

I'm running into the same issue, haven't found a definite solution yet but DjangoChannelsGraphqlWsl seems to be promising. What caught my eye was:

• All GraphQL "resolvers" run in a threadpool so they never block the server itself and may communicate with database or perform other blocking tasks.
• Resolvers (including subscription's subscribe & publish) can be represented both as synchronous or asynchronous (async def) methods.

[Suor]

I guess you should pass a non-default executor to the view:

from graphql.execution.executors.asyncio import AsyncioExecutor
                         
graphql_view = GraphQLView.as_view(executor=AsyncioExecutor())

[Suor]

Apparently setting executor doesn't work. It would be nice to have this since Django 3.1 already have async views.

[blazing-gig]

It would be really nice to have a separate AsyncGraphQLView similar to this one from the Strawberry library. Any suggestions or alternatives would be really appreciated as well.

[fabienheureux]

@jkimbo created a repl to showcase dataloaders with graphene 3 / async resolvers.
They need a async views that is in the same repl https://replit.com/@jkimbo/Graphene-async#main/schema.py

[fabienheureux]

If someone want to give a hand, I started a PR to add an async view to graphene_django #1256

[sheenobu]

NOTE: self is busted. maybe it's setattr(schema, "execute", partialmethod(schema.execute, executor=GeventExecutor))?

would something like this work or does the view itself have to be async? maybe it would work with gevent but not Asyncio executors?

from graphql.execution.executors.asyncio import AsyncioExecutor
from funcy import func_partial

schema = graphene.Schema(query=Query, mutation=Mutation)                                                    
setattr(schema, 'execute', func_partial(schema.execute, executor=AsyncioExecutor))

[lejansenampeers]

strawberry django plug-in comes with async. Worked without any problems for me:

https://strawberry.rocks/docs/integrations/django#async-django