在 FastAPI 中，不能使用解析器异步等待

Question

我正在将石墨烯与 fastapi 一起使用，但无法使简单的异步解析器正常工作。感谢您的帮助。

class Query(graphene.ObjectType):
    hello = graphene.String(name=graphene.String(default_value=""))

    async def resolve_hello(self, info, name):
      image_url = "my_url"
      async with httpx.AsyncClient() as client:
        response = await client.get(image_url)
        # img_bytes = await get_bytes(request.query_params["url"])
        // do stuff

错误是

{
  "data": {
    "hello": null
  },
  "errors": [
    {
      "message": "There is no current event loop in thread 'ThreadPoolExecutor-1_0'."
    }
  ]
}

Answer 1

TL;DR

将 GraphQL 端点添加到 app 对象时，将其替换为

app = FastAPI()
app.add_route("/", GraphQLApp(schema=graphene.Schema(query=Query),
    executor_class=AsyncioExecutor))

完整答案

好的，所以我在检查文档并将它们与您的代码进行比较后发现了您的问题。

这不是 httpx 的问题，而是您如何使用 GraphQL 的问题。

正如您在 Fastapi Docs, the resolve_hello function is declared with a traditional def instead of async def. This makes a difference for the GraphQL engine when running because it calls the function, but does not await it, instead, it gets the coroutine object and goes on with returning the result. Since the result is a coroutine it does not work and throws the exception you posted. The solution lays in the Starlette Docs 中看到的那样，它将 asyncio 引擎作为参数传递给运行 async GraphQL 查询。

这很棘手，因为 fastAPI 基于 Starlette，因此您必须阅读这两个文档以备不时之需。

如果问题仍然存在，请告诉我。

Answer 2

因为 Graphene (2) 不再使用 AsyncioExecutor

    app = FastAPI()
    app.add_route("/", GraphQLApp(schema=graphene.Schema(query=Query),
    executor_class=AsyncioExecutor))

我建议使用 Strawbery or starlette-graphene3 如果你想使用 pydantic

pip install starlette-graphene3

如果需要升级你的 pip

    import asyncio

    import graphene
    import pydantic
    from fastapi import FastAPI
    from starlette.applications import Starlette
    from starlette_graphene3 import GraphQLApp, make_graphiql_handler
    from graphene_pydantic import PydanticObjectType


    class PersonModel(pydantic.BaseModel):
        first_name: str
        last_name: str


    class Person(PydanticObjectType):
        class Meta:
            model = PersonModel


    async def wait_for_me(wo):
        return wo

    class Query(graphene.ObjectType):
        people = graphene.List(Person)

        async def resolve_people(root, info):
            mmd = await wait_for_me("jose")
            return [PersonModel(first_name=mmd, last_name="Smith")]


    class Subscription(graphene.ObjectType):
        count = graphene.Int(upto=graphene.Int())

        async def subscribe_count(root, info, upto=3):
            for i in range(upto):
                yield i
                # await asyncio.sleep(1)


    app = FastAPI()


    @app.get("/hello")
    def read_root():
        return {"Hello": "World"}



    schema = graphene.Schema(query=Query, subscription=Subscription)


    app.mount("/", GraphQLApp(schema, on_get=make_graphiql_handler()))  # Graphiql IDE
    # app.mount("/", GraphQLApp(schema, on_get=make_playground_handler()))  # Playground IDE
    # app.mount("/", GraphQLApp(schema)) # no IDE

在 FastAPI 中，不能使用解析器异步等待

in FastAPI, can not use resolver async await

graphene-python

fastapi