FastAPI graphql订阅

Question

我尝试在FastAPI上实现简单的graphql订阅。

根据文档，但它不起作用

import asyncio

import graphene
from fastapi import FastAPI
from starlette.graphql import GraphQLApp
from graphql.execution.executors.asyncio import AsyncioExecutor
from starlette.websockets import WebSocket


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

    async def resolve_hello(self,info,name):
        return "Hello " + name

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


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

我谷歌了一下，发现我可能需要像Sanic或Aiohttp那样实现订阅服务器

我试过了，但还不起作用

from graphql_ws.websockets_lib import WsLibSubscriptionServer

subscription_server = WsLibSubscriptionServer(schema)

@app.websocket("/subscriptions")
async def subscriptions(websocket: WebSocket):
    await subscription_server.handle(websocket)
    return websocket

错误接收：

 return self.ws.open is False
AttributeError: 'WebSocket' object has no attribute 'open'

我做错了什么，该如何解决？谢谢。

Answer 1

我想通了

以下是可行的解决方案

import asyncio

import graphene
import pydantic
from fastapi import FastAPI
from starlette_graphene3 import GraphQLApp, make_playground_handler
from graphene_pydantic import PydanticObjectType


class Subscription(graphene.ObjectType):
    count = graphene.String()

    async def subscribe_count(root, info):
        for i in range(300):
            yield f"{i}"
            await asyncio.sleep(1)

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


app.mount("/", GraphQLApp(schema, on_get=make_playground_handler()))