Unable to run query with async resolver in FastAPI/starlette-graphene3

Question

I am trying to implement a GraphQL app served via FastAPI. The graphql query schema has a mix of async and sync resolvers, as I am trying to move away from GraphQL2. Trying to use fastAPI, starlette-graphene3, graphene, and graphql-core. When I try to run a query like:

query {{
  thread(id: "123") {{
    id
  }}
}}

I get weird errors like: TypeError: Cannot return null for non-nullable field Thread.id.. Additionally, when remote debugging the app, the debugger never steps into the resolver functions. If I test by attempting to switch some async resolvers to sync, the resolvers seem to run out of order resulting in other bizarre errors.

class Query(graphene.ObjectType):
   me = graphene.Field(User)
   thread = graphene.Field(Thread, thread_id=graphene.String(required=True, name="id"))
   ...
   @staticmethod
   async def resolve_thread(parent, info: GrapheneInfo, thread_id):
        return await info.context.thread_loader.load(thread_id)

   @staticmethod
   def resolve_some_other_field(parent, info: GrapheneInfo, field_id):
        ...
   ...

The Thread object:

class Thread(graphene.ObjectType):
    id = graphene.ID(required=True)
    topic = graphene.String(required=True)
    post = graphene.Field(lambda: Post, required=True)
    
    @staticmethod
    async def resolve_thread(thread: ThreadModel, info: GrapheneInfo, **kwargs):
        return await info.context.post_loader.load(thread.post_id)
    ...

The ThreadLoader class:

from typing import Optional

from aiodataloader import DataLoader
...

class ThreadLoader(DataLoader[str, Optional[ThreadModel]]):
    thread_service: ThreadService

    def __init__(self, thread_service: ThreadService):
        super(ThreadLoader, self).__init__()
        self.thread_service = thread_service

    async def batch_load_fn(self, keys: list[str]) -> list[Optional[ThreadModel]]:
        # Thread service function is not async
        threads = self.thread_service.get_threads(keys)

The graphqlApp initialization:

def schema():
    # Set up the schema to our root level queries and mutators
    return Schema(
        query=Query,
        types=[],
    )
...
def setup_graph_ql_app(get_context: Callable[[], Context]) -> GraphQLApp:
    return GraphQLApp(
        schema=schema(),
        on_get=get_graphiql_handler(),
        context_value=_get_context_callable(get_context),
        middleware=[
            ...
        ],
    )

I add this to my fastAPI app like so:

fast_api_app = FastAPI()
graphql_app: GraphQLApp = setup_graph_ql_app(get_context=context_callable)
app.add_route("/v1/graphql", graphql_app)

The dependencies that I've defined are:

graphene = "^3.2"
starlette-graphene3 = "^0.6.0"
graphql-core = "^3.2.0"
fastapi = "^0.109.0"
uvloop = "^0.19.0"
asyncio = "^3.4.3"
aiodataloader = "^0.4.0"

Looking at the documentation here: https://docs.graphene-python.org/_/downloads/sqlalchemy/en/latest/pdf/ and https://docs.graphene-python.org/en/latest/execution/dataloader/#dataloader seems to match the classes above. Is there something I am missing or that I am doing incorrectly?

Answer 1

The problem ended up being in a middleware function on the GraphQLApp. The problem was that one of the resolve functions was not correctly handling the async return:

    if isinstance(next, Awaitable):
        return await next(root, info, **kwargs)
    return next(root, info, **kwargs)

In Graphene 3, the next function can be async or a regular function. Additionally, it is not the next arg that needs to be checked it is the return value of invoking next.

    return_value = next(root, info, **kwargs)
    if inspect.isawaitable(return_value):
        return await return_value
    return return_value