Skip to main content

Websocket backend for GraphQL subscriptions.

Supports the following application servers:

Python 3 application servers, using asyncio:

Python 2 application servers:

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:

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:

  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'
# 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

Release files for graphql-ws 0.4.4

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for graphql-ws 0.4.4
File Size Uploaded
graphql-ws-0.4.4.tar.gz 47.9 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for graphql-ws 0.4.4
File Interpreter ABI Platform
graphql_ws-0.4.4-py2.py3-none-any.whl Python 3, Python 2 none any Details

Total release size: 65.2 kB

Release files / graphql-ws-0.4.4.tar.gz

Download URL graphql-ws-0.4.4.tar.gz
Size 47.9 kB
Tags Source
SHA-256 checksum
How to use checksums
2ad38db70f37964f54d7eb3e2ede86dbe3f2a1ed7ea0a9f9a3b8b17162a22852
BLAKE2b-256 checksum
How to use checksums
c41f7987d85bdd997e01637d9135cf9896131614f3322f1cd40b0e125633c894
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/3.4.2 importlib_metadata/4.6.4 pkginfo/1.7.1 requests/2.26.0 requests-toolbelt/0.9.1 tqdm/4.62.2 CPython/3.7.11

Release files / graphql_ws-0.4.4-py2.py3-none-any.whl

Download URL graphql_ws-0.4.4-py2.py3-none-any.whl
Size 17.3 kB
Tags Python 2 Python 3
SHA-256 checksum
How to use checksums
b6f4c9f6968feba80762354068a2a36538a48ac72e4253971be43e0cba020506
BLAKE2b-256 checksum
How to use checksums
9719fd64b7972099032bfa88358ad41e95c3348a73661a23727ba2981d43d2af
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/3.4.2 importlib_metadata/4.6.4 pkginfo/1.7.1 requests/2.26.0 requests-toolbelt/0.9.1 tqdm/4.62.2 CPython/3.7.11

Release history Release notifications | RSS feed

This release

0.4.4 This release

2 release files

0.4.3

2 release files

0.3.1

2 release files

0.3.0

2 release files

0.2.0

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page