Skip to content

using async resolvers for third-party api calls #870

Open
@Schayik

Description

@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!

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions