GraphQL WS

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)

Installation instructions

For instaling graphql-ws, just run this command in your shell

pip install graphql-ws

Examples

Python 3 servers

Create a subscribable schema like this:

importasyncioimportgrapheneclassQuery(graphene.ObjectType):
hello=graphene.String()
@staticmethoddefresolve_hello(obj, info, **kwargs):
return"world"classSubscription(graphene.ObjectType):
count_seconds=graphene.Float(up_to=graphene.Int())
asyncdefresolve_count_seconds(root, info, up_to):
foriinrange(up_to):
yieldiawaitasyncio.sleep(1.)
yieldup_toschema=graphene.Schema(query=Query, subscription=Subscription)

aiohttp

Then just plug into your aiohttp server.

fromgraphql_ws.aiohttpimportAiohttpSubscriptionServerfrom .schemaimportschemasubscription_server=AiohttpSubscriptionServer(schema)
asyncdefsubscriptions(request):
ws=web.WebSocketResponse(protocols=('graphql-ws',))
awaitws.prepare(request)
awaitsubscription_server.handle(ws)
returnwsapp=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

websockets compatible servers

Works with any framework that uses the websockets library for its websocket implementation. For this example, plug in your Sanic server.

fromgraphql_ws.websockets_libimportWsLibSubscriptionServerfrom . importschemaapp=Sanic(__name__)
subscription_server=WsLibSubscriptionServer(schema)
@app.websocket('/subscriptions', subprotocols=['graphql-ws'])asyncdefsubscriptions(request, ws):
awaitsubscription_server.handle(ws)
returnwsapp.run(host="0.0.0.0", port=8000)

Django v2+

Django Channels 2

Set up with Django Channels just takes three steps:

1. Install the apps
2. Set up your schema
3. 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'# orASGI_APPLICATION='graphql_ws.django.routing.auth_application'

Run ./manage.py runserver and go to http://localhost:8000/graphql to test!

Python 2 servers

Create a subscribable schema like this:

importgraphenefromrximportObservableclassQuery(graphene.ObjectType):
hello=graphene.String()
@staticmethoddefresolve_hello(obj, info, **kwargs):
return"world"classSubscription(graphene.ObjectType):
count_seconds=graphene.Float(up_to=graphene.Int())
asyncdefresolve_count_seconds(root, info, up_to=5):
returnObservable.interval(1000)\
.map(lambdai: "{0}".format(i))\
.take_while(lambdai: int(i) <=up_to)
schema=graphene.Schema(query=Query, subscription=Subscription)

Gevent compatible servers

Then just plug into your Gevent server, for example, Flask:

fromflask_socketsimportSocketsfromgraphql_ws.geventimportGeventSubscriptionServerfromschemaimportschemasubscription_server=GeventSubscriptionServer(schema)
app.app_protocol=lambdaenviron_path_info: 'graphql-ws'@sockets.route('/subscriptions')defecho_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

Django v1.x

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

fromchannels.routingimportroute_classfromgraphql_ws.django_channelsimportGraphQLSubscriptionConsumerchannel_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