Websocket backend for GraphQL subscriptions.
Supports the following application servers:
Python 3 application servers, using asyncio:
- aiohttp
- websockets compatible servers such as Sanic (via websockets library)
- Django v2+
Python 2 application servers:
- Gevent compatible servers such as Flask
- Django v1.x (via channels v1.x)
For instaling graphql-ws, just run this command in your shell
pip install graphql-wsCreate a subscribable schema like this:
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)Then just plug into your aiohttp server.
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)You can see a full example here: https://github.com/graphql-python/graphql-ws/tree/master/examples/aiohttp
Works with any framework that uses the websockets library for its websocket implementation. For this example, plug in your Sanic server.
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)Set up with Django Channels just takes three steps:
- Install the apps
- Set up your schema
- Configure the channels router application
First pip install channels and it to your INSTALLED_APPS. If you
want graphiQL, install the graphql_ws.django app before
graphene_django to serve a graphiQL template that will work with
websockets:
INSTALLED_APPS = [
"channels",
"graphql_ws.django",
"graphene_django",
# ...
]Point to your schema in Django settings:
GRAPHENE = {
'SCHEMA': 'yourproject.schema.schema'
}Finally, you can set up channels routing yourself (maybe using
graphql_ws.django.routing.websocket_urlpatterns in your
URLRouter), or you can just use one of the preset channels
applications:
ASGI_APPLICATION = 'graphql_ws.django.routing.application'
# or
ASGI_APPLICATION = 'graphql_ws.django.routing.auth_application'Run ./manage.py runserver and go to
http://localhost:8000/graphql to test!
Create a subscribable schema like this:
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)Then just plug into your Gevent server, for example, 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 []You can see a full example here: https://github.com/graphql-python/graphql-ws/tree/master/examples/flask_gevent
For Django v1.x and Django Channels v1.x, setup your schema in settings.py
GRAPHENE = {
'SCHEMA': 'yourproject.schema.schema'
}Then pip install "channels<1" and it to your django apps, adding the
following to your settings.py
CHANNELS_WS_PROTOCOLS = ["graphql-ws", ]
CHANNEL_LAYERS = {
"default": {
"BACKEND": "asgiref.inmemory.ChannelLayer",
"ROUTING": "django_subscriptions.urls.channel_routing",
},
}And finally add the channel routes
from channels.routing import route_class
from graphql_ws.django_channels import GraphQLSubscriptionConsumer
channel_routing = [
route_class(GraphQLSubscriptionConsumer, path=r"^/subscriptions"),
]You can see a full example here: https://github.com/graphql-python/graphql-ws/tree/master/examples/django_subscriptions