GitHub - graphql-python/graphql-ws: GraphQL websockets

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

Examples

Python 3 servers

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

aiohttp

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

websockets compatible servers

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)

Django v2+

Django Channels 2

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!

Python 2 servers

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)

Gevent compatible servers

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

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

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