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

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

graphql-ws-0.4.4.tar.gz (47.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

graphql_ws-0.4.4-py2.py3-none-any.whl (17.3 kB view details)

Uploaded Python 2Python 3

File details

Details for the file graphql-ws-0.4.4.tar.gz.

File metadata

  • Download URL: graphql-ws-0.4.4.tar.gz
  • Upload date:
  • Size: 47.9 kB
  • Tags: Source
  • Uploaded using 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

File hashes

Hashes for graphql-ws-0.4.4.tar.gz
Algorithm Hash digest
SHA256 2ad38db70f37964f54d7eb3e2ede86dbe3f2a1ed7ea0a9f9a3b8b17162a22852
MD5 2e7a73123ac41cf81dd33785865da1c4
BLAKE2b-256 c41f7987d85bdd997e01637d9135cf9896131614f3322f1cd40b0e125633c894

See more details on using hashes here.

File details

Details for the file graphql_ws-0.4.4-py2.py3-none-any.whl.

File metadata

  • Download URL: graphql_ws-0.4.4-py2.py3-none-any.whl
  • Upload date:
  • Size: 17.3 kB
  • Tags: Python 2, Python 3
  • Uploaded using 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

File hashes

Hashes for graphql_ws-0.4.4-py2.py3-none-any.whl
Algorithm Hash digest
SHA256 b6f4c9f6968feba80762354068a2a36538a48ac72e4253971be43e0cba020506
MD5 ece19d51f24c4d55e567823b55983cd8
BLAKE2b-256 9719fd64b7972099032bfa88358ad41e95c3348a73661a23727ba2981d43d2af

See more details on using hashes here.

Release history Release notifications | RSS feed

This release

0.4.4 This release

2 files

0.4.3

2 files

0.3.1

2 files

0.3.0

2 files

0.2.0

2 files

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page