GraphQL订阅的Websocket后端
项目描述
GraphQL订阅的Websocket后端。
支持以下应用服务器
Python 3应用服务器，使用asyncio
与websockets兼容的服务器，例如Sanic（通过websockets库）
Python 2应用服务器
与Gevent兼容的服务器，例如Flask
安装说明
要安装graphql-ws，只需在您的shell中运行此命令
pip install graphql-ws示例
Python 3服务器
创建一个可订阅的schema，如下所示
import asyncio
import graphene
class Query(graphene.ObjectType):
hello = graphene.String()
@staticmethod
def resolve_hello(obj, info, **kwargs):
return "world"
class Subscription(graphene.ObjectType):
count_seconds = graphene.Float(up_to=graphene.Int())
async def resolve_count_seconds(root, info, up_to):
for i in range(up_to):
yield i
await asyncio.sleep(1.)
yield up_to
schema = graphene.Schema(query=Query, subscription=Subscription)aiohttp
然后只需将其连接到您的aiohttp服务器。
from graphql_ws.aiohttp import AiohttpSubscriptionServer
from .schema import schema
subscription_server = AiohttpSubscriptionServer(schema)
async def subscriptions(request):
ws = web.WebSocketResponse(protocols=('graphql-ws',))
await ws.prepare(request)
await subscription_server.handle(ws)
return ws
app = web.Application()
app.router.add_get('/subscriptions', subscriptions)
web.run_app(app, port=8000)您可以在这里看到完整的示例： https://github.com/graphql-python/graphql-ws/tree/master/examples/aiohttp
与websockets兼容的服务器
支持任何使用WebSocket库实现WebSocket的框架。在此示例中，请插入您的Sanic服务器。
from graphql_ws.websockets_lib import WsLibSubscriptionServer
from . import schema
app = Sanic(__name__)
subscription_server = WsLibSubscriptionServer(schema)
@app.websocket('/subscriptions', subprotocols=['graphql-ws'])
async def subscriptions(request, ws):
await subscription_server.handle(ws)
return ws
app.run(host="0.0.0.0", port=8000)Django v2+
Django Channels 2
使用Django Channels只需三个步骤
安装应用程序
设置您的模式
配置通道路由器应用程序
首先 pip install channels 并将其添加到 INSTALLED_APPS。如果您想使用graphiQL，请在安装 graphene_django 之前安装 graphql_ws.django 应用程序，以提供与WebSocket兼容的graphiQL模板
INSTALLED_APPS = [
"channels",
"graphql_ws.django",
"graphene_django",
# ...
]在Django设置中指向您的模式
GRAPHENE = {
'SCHEMA': 'yourproject.schema.schema'
}最后，您可以自己设置通道路由（可能使用 graphql_ws.django.routing.websocket_urlpatterns 在您的 URLRouter 中），或者您可以直接使用预置的通道应用程序之一
ASGI_APPLICATION = 'graphql_ws.django.routing.application'
# or
ASGI_APPLICATION = 'graphql_ws.django.routing.auth_application'运行 ./manage.py runserver 并转到 https://:8000/graphql 进行测试！
Python 2服务器
创建一个可订阅的schema，如下所示
import graphene
from rx import Observable
class Query(graphene.ObjectType):
hello = graphene.String()
@staticmethod
def resolve_hello(obj, info, **kwargs):
return "world"
class Subscription(graphene.ObjectType):
count_seconds = graphene.Float(up_to=graphene.Int())
async def resolve_count_seconds(root, info, up_to=5):
return Observable.interval(1000)\
.map(lambda i: "{0}".format(i))\
.take_while(lambda i: int(i) <= up_to)
schema = graphene.Schema(query=Query, subscription=Subscription)兼容Gevent的服务器
然后只需将其连接到您的Gevent服务器，例如，Flask
from flask_sockets import Sockets
from graphql_ws.gevent import GeventSubscriptionServer
from schema import schema
subscription_server = GeventSubscriptionServer(schema)
app.app_protocol = lambda environ_path_info: 'graphql-ws'
@sockets.route('/subscriptions')
def echo_socket(ws):
subscription_server.handle(ws)
return []您可以在以下位置查看完整示例： https://github.com/graphql-python/graphql-ws/tree/master/examples/flask_gevent
Django v1.x
对于Django v1.x和Django Channels v1.x，在 settings.py 中设置您的模式
GRAPHENE = {
'SCHEMA': 'yourproject.schema.schema'
}然后 pip install "channels<1" 并将其添加到您的django应用程序中，将以下内容添加到您的 settings.py
CHANNELS_WS_PROTOCOLS = ["graphql-ws", ]
CHANNEL_LAYERS = {
"default": {
"BACKEND": "asgiref.inmemory.ChannelLayer",
"ROUTING": "django_subscriptions.urls.channel_routing",
},
}最后添加通道路由
from channels.routing import route_class
from graphql_ws.django_channels import GraphQLSubscriptionConsumer
channel_routing = [
route_class(GraphQLSubscriptionConsumer, path=r"^/subscriptions"),
]您可以在以下位置查看完整示例： https://github.com/graphql-python/graphql-ws/tree/master/examples/django_subscriptions
下载文件
下载适合您平台的应用程序。如果您不确定要选择哪个，请了解更多关于 安装包 的信息。
