Metadata-Version: 2.1
Name: graphql-ws-subs
Version: 0.0.2
Summary: Websocket server for GraphQL subscriptions that includes subscriptions implementation from hballard
Home-page: https://github.com/cjayroe/graphql-ws
Author: Syrus Akbary
Author-email: me@syrusakbary.com
License: MIT
Keywords: graphql,subscriptions,graphene,websockets
Platform: UNKNOWN
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Natural Language :: English
Classifier: Programming Language :: Python :: 2
Classifier: Programming Language :: Python :: 2.6
Classifier: Programming Language :: Python :: 2.7
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.3
Classifier: Programming Language :: Python :: 3.4
Classifier: Programming Language :: Python :: 3.5
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Requires-Dist: graphql-core (==2.*)
Provides-Extra: dev
Requires-Dist: flake8 (<4,>=3.7) ; extra == 'dev'
Requires-Dist: tox (<4,>=3) ; extra == 'dev'
Requires-Dist: Sphinx (<2,>=1.8) ; extra == 'dev'
Provides-Extra: maintainer
Requires-Dist: bumpversion (>=0.5.3) ; extra == 'maintainer'
Requires-Dist: wheel (>=0.33.6) ; extra == 'maintainer'
Requires-Dist: PyYAML (<6,>=5.3) ; extra == 'maintainer'
Provides-Extra: test
Requires-Dist: pytest-cov ; extra == 'test'
Requires-Dist: graphene (<3,>=2.0) ; extra == 'test'
Requires-Dist: gevent ; extra == 'test'
Requires-Dist: graphene (>=2.0) ; extra == 'test'
Requires-Dist: graphene-django ; extra == 'test'
Requires-Dist: pytest (<5,>=4) ; (python_version < "3") and extra == 'test'
Requires-Dist: mock ; (python_version < "3") and extra == 'test'
Requires-Dist: django (==1.11.*) ; (python_version < "3") and extra == 'test'
Requires-Dist: channels (==1.*) ; (python_version < "3") and extra == 'test'
Requires-Dist: pytest (>5) ; (python_version >= "3") and extra == 'test'
Requires-Dist: django (==2.*) ; (python_version >= "3") and extra == 'test'
Requires-Dist: channels (==2.*) ; (python_version >= "3") and extra == 'test'
Requires-Dist: pytest-asyncio ; (python_version >= "3.4") and extra == 'test'
Requires-Dist: aiohttp ; (python_version >= "3.5") and extra == 'test'

GraphQL WS
==========

Websocket server for GraphQL subscriptions.

Currently supports:

* `aiohttp <https://github.com/graphql-python/graphql-ws#aiohttp>`__
* `Gevent <https://github.com/graphql-python/graphql-ws#gevent>`__
* Sanic (uses `websockets <https://github.com/aaugustin/websockets/>`__
  library)

Installation instructions
=========================

For instaling graphql-ws, just run this command in your shell

.. code:: bash

    pip install graphql-ws

Examples
--------

aiohttp
~~~~~~~

For setting up, just plug into your aiohttp server.

.. code:: python

    from graphql_ws.aiohttp import AiohttpSubscriptionServer


    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)

Sanic
~~~~~

Works with any framework that uses the websockets library for it’s
websocket implementation. For this example, plug in your Sanic server.

.. code:: python

    from graphql_ws.websockets_lib import WsLibSubscriptionServer


    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)

And then, plug into a subscribable schema:

.. code:: python

    import asyncio
    import graphene


    class Query(graphene.ObjectType):
        base = graphene.String()


    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)

You can see a full example here:
https://github.com/graphql-python/graphql-ws/tree/master/examples/aiohttp

Gevent
~~~~~~

For setting up, just plug into your Gevent server.

.. code:: python

    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 []

And then, plug into a subscribable schema:

.. code:: python

    import graphene
    from rx import Observable


    class Query(graphene.ObjectType):
        base = graphene.String()


    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)

You can see a full example here:
https://github.com/graphql-python/graphql-ws/tree/master/examples/flask_gevent

Django Channels
~~~~~~~~~~~~~~~

First ``pip install channels`` and it to your django apps

Then add the following to your settings.py

.. code:: python

    CHANNELS_WS_PROTOCOLS = ["graphql-ws", ]
    CHANNEL_LAYERS = {
        "default": {
            "BACKEND": "asgiref.inmemory.ChannelLayer",
            "ROUTING": "django_subscriptions.urls.channel_routing",
        },

    }

Setup your graphql schema

.. code:: python

    import graphene
    from rx import Observable


    class Query(graphene.ObjectType):
        hello = graphene.String()

        def resolve_hello(self, info, **kwargs):
            return 'world'

    class Subscription(graphene.ObjectType):

        count_seconds = graphene.Int(up_to=graphene.Int())


        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
    )

Setup your schema in settings.py

.. code:: python

    GRAPHENE = {
        'SCHEMA': 'path.to.schema'
    }

and finally add the channel routes

.. code:: python

    from channels.routing import route_class
    from graphql_ws.django_channels import GraphQLSubscriptionConsumer

    channel_routing = [
        route_class(GraphQLSubscriptionConsumer, path=r"^/subscriptions"),
    ]


