Websocket backend for GraphQL subscriptions.
Supports the following application servers:
Python 3 application servers, using asyncio:
websockets compatible servers such as Sanic (via websockets library)
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:
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
Metadata
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)
| File | Size | Uploaded | Provenance | Actions |
|---|---|---|---|---|
| graphql-ws-0.4.4.tar.gz | 47.9 kB | Without provenance | Details for graphql-ws-0.4.4.tar.gz |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Actions Reset |
|---|---|---|---|---|
| graphql_ws-0.4.4-py2.py3-none-any.whl | Python 3, Python 2 | none | any | Details for graphql_ws-0.4.4-py2.py3-none-any.whl |
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
|
Security for graphql-ws 0.4.4
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.