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?
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