无法在 FastAPI/starlette-graphene3 中使用异步解析器运行查询

Question

我正在尝试实现一个通过 FastAPI 提供服务的 GraphQL 应用程序。 graphql 查询模式混合了

async

和

sync

解析器，因为我正在尝试摆脱 GraphQL2。尝试使用 fastAPI、starlette-graphene3、graphene 和 graphql-core。当我尝试运行如下查询时：

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

我收到奇怪的错误，例如：

TypeError: Cannot return null for non-nullable field Thread.id.

。此外，在远程调试应用程序时，调试器永远不会进入解析器函数。如果我通过尝试将一些异步解析器切换为同步来进行测试，解析器似乎会乱序运行，从而导致其他奇怪的错误。

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):
        ...
   ...

Thread

对象：

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

ThreadLoader

班：

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)

graphqlApp

初始化：

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=[
            ...
        ],
    )

我将其添加到我的 fastAPI 应用程序中，如下所示：

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

我定义的依赖项是：

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"

查看此处的文档：https://docs.graphene-python.org/_/downloads/sqlalchemy/en/latest/pdf/和https://docs.graphene-python.org/en/latest /execution/dataloader/#dataloader 似乎与上面的类匹配。我是否遗漏了什么或我做错了什么？

Answer 1

问题最终出在

GraphQLApp

上的中间件功能上。问题是其中一个解析函数没有正确处理异步返回：

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

在石墨烯3中，

next

函数可以是

async

或常规函数。另外，需要检查的不是

next

参数，而是调用 next 的返回值。

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