Я пытаюсь реализовать приложение GraphQL, обслуживаемое через FastAPI. Схема запроса Graphql содержит смесь преобразователей async и sync, поскольку я пытаюсь отойти от GraphQL2. Пытаюсь использовать fastAPI, starlette-graphene3, графен и графл-ядро. Когда я пытаюсь выполнить запрос типа:
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/ late/execution/dataloader/#dataloader кажется соответствует приведенным выше классам. Что-то мне не хватает или я делаю неправильно?





Проблема оказалась в функции промежуточного программного обеспечения GraphQLApp. Проблема заключалась в том, что одна из функций разрешения неправильно обрабатывала асинхронный возврат:
if isinstance(next, Awaitable):
return await next(root, info, **kwargs)
return next(root, info, **kwargs)
В Graphene 3 функция next может быть async или обычной функцией. Кроме того, необходимо проверять не аргумент next, а возвращаемое значение вызова next.
return_value = next(root, info, **kwargs)
if inspect.isawaitable(return_value):
return await return_value
return return_value