• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号
登陆 注册
描述文字

graphql-python/graphql-ws: GraphQL websockets

原作者: [db:作者] 来自: 网络 收藏 邀请

开源软件名称（OpenSource Name）：

graphql-python/graphql-ws

开源软件地址(OpenSource Url)：

https://github.com/graphql-python/graphql-ws

开源编程语言(OpenSource Language)：

Python 87.0%

开源软件介绍(OpenSource Introduction)：

GraphQL WS

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

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
vendure-ecommerce/vendure: A headless GraphQL ecommerce framework for the modern ...发布时间：2022-06-22
下一篇：
adrianhajdin/project_graphql_blog: With featured and recent posts, categories. f ...发布时间：2022-06-22
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap