Graphene real-time subscriptions and Apollo client graphql integration

[hballard]

Hello @syrusakbary.

Thanks for all your hard work on graphene and graphql-python. Awesome library!!

I posted this on #393 earlier this week...reposting here so it's easier to discover.

I implemented a port of the apollo graphql subscriptions modules (graphql-subscriptions and subscriptions-transport-ws) for graphene / python. They work w/ apollo-client.

It is here.

Same basic api as the Apollo modules. It is still very rough...but works so far, based on my limited internal testing. Uses redis-py, gevent-websockets, and syrusakbary/promises. I was going to add a simple example app, setup.py for easier install, and more info to the readme w/ the API, in the next few days. A brief example is below. Only works on python2 for now. My plan is to start working on tests as well. I figured I'd go ahead and share in this early stage in case anybody is interested...

I'm very new to open source, so any critiques or pull requests are welcome.

Simple example:

Server (using Flask and Flask-Sockets):

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_sockets import Sockets

from .subscription_manager import SubscriptionManager, RedisPubsub
from .subscription_transport_ws import ApolloSubscriptionServer

app = Flask(__name__)
sockets = Sockets(app)
pubsub = RedisPubsub()
schema = graphene.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription
)
subscription_mgr = SubscriptionManager(schema, pubsub)

@sockets.route('/socket')
def socket_channel(websocket):
    subscription_server = ApolloSubscriptionServer(subscription_mgr, websocket)
    subscription_server.handle()
    return []

if __name__ == "__main__":
    from geventwebsocket import WebSocketServer

    server = WebSocketServer(('', 5000), app)
    print '  Serving at host 0.0.0.0:5000...\n'
    server.serve_forever()

Of course on the server you have to "publish" each time you have a mutation (in this case to a redis channel). That could look something like this (using graphene / sql-alchemy):

class Subscription(graphene.ObjectType):
    users = graphene_sqlalchemy.SQLAlchemyConnectionField(
        User,
        active=graphene.Boolean()
    )

    def resolve_users(self, args, context, info):
        query = User.get_query(context)
        return query.filter_by(id=info.root_value.get('id'))

class AddUser(graphene.ClientIDMutation):

    class Input:
        username = graphene.String(required=True)
        email = graphene.String()

    ok = graphene.Boolean()
    user = graphene.Field(lambda: User)

    @classmethod
    def mutate_and_get_payload(cls, args, context, info):
        _input = args.copy()
        del _input['clientMutationId']
        new_user = UserModel(**_input)
        db.session.add(new_user)
        db.session.commit()
        ok = True
        if pubsub.subscriptions:
            pubsub.publish('users', new_user.as_dict())
        return AddUser(ok=ok, user=new_user)

Client (using react-apollo client):

import React from 'react'
import ReactDOM from 'react-dom'
import { graphql, ApolloProvider } from 'react-apollo'
import gql from 'graphql-tag'
import ApolloClient, { createNetworkInterface } from 'apollo-client'
import { SubscriptionClient, addGraphQLSubscriptions } from 'subscriptions-transport-ws'

import ChatApp from './screens/ChatApp'
import ListBox from '../components/ListBox'

const SUBSCRIPTION_QUERY = gql`
  subscription newUsers {
    users(active: true) {
      edges {
        node {
          id
          username
        }
      }
    }
  }
`

const LIST_BOX_QUERY = gql`
  query AllUsers {
    users(active: true) {
      edges {
        node {
          id
          username
        }
      }
    }
  }
`

class ChatListBox extends React.Component {

  componentWillReceiveProps(newProps) {
    if (!newProps.data.loading) {
      if (this.subscription) {
        return
      }
      this.subscription = newProps.data.subscribeToMore({
        document: SUBSCRIPTION_QUERY,
        updateQuery: (previousResult, {subscriptionData}) => {
          const newUser = subscriptionData.data.users.edges
          const newResult = {
            users: {
              edges: [
                ...previousResult.users.edges,
                ...newUser
              ]
            }
          }
          return newResult
        },
        onError: (err) => console.error(err)
      })
    }
  }

  render() {
    return <ListBox data={this.props.data} />
  }
}

const ChatListBoxWithData = graphql(LIST_BOX_QUERY)(ChatListBox)

export default ChatListBoxWithData

const networkInterface = createNetworkInterface({
  uri: 'http://localhost:5000/graphql'
})

const wsClient = new SubscriptionClient(`ws://localhost:5000/socket`, {
  reconnect: true
})

const networkInterfaceWithSubscriptions = addGraphQLSubscriptions(
  networkInterface,
  wsClient,
)

const client = new ApolloClient({
  dataIdFromObject: o => o.id,
  networkInterface: networkInterfaceWithSubscriptions
})

ReactDOM.render(
  <ApolloProvider client={client}>
    <ChatApp />
  </ApolloProvider>,
  document.getElementById('root')
)

[syrusakbary]

This is awesome!

I will take a closer look next week and provide some feedback then :)

[jeffreybrowning]

@syrusakbary Thoughts?

[dudanogueira]

Nice!!!

[ignatevdev]

@syrusakbary Any update on this?

[syrusakbary]

I took a look in the implementation and besides some small nits (like Apollo* naming) it looked good!

I'm waiting to few things:

• Having a subscriptions integration with Django so we assure that the subscriptions structure is abstracted in a scalable way.
• The formal subscription RFC in the GraphQL spec to be merged Spec: Formal Subscriptions Definition graphql/graphql-spec#305

[hballard]

Thanks for the comments @syrusakbary . I think your grapqhl / graphene / promises libraries are amazing...any constructive criticism you have I'm happy to hear. I only used the "Apollo.." naming convention because my implementation was initially based on their (Apollo's) graphql subscription transport protocol. But, as the final spec should be merged soon, I'm happy to drop that convention, since it only affects the main subscription transport class. I've been tied up the last couple months, since I published this, so I haven't been able to devote much time to improving it. Some priorities in the near term for me:

• Finish up tests. I'm about done w/ the last bit (focused on the transport module) and should be publishing them in the next few days.
• Add Python 3 support
• Add Django support
• Finish simple example chat app
• Abstract the concurrency executor (uses Gevent now...but obviously with python >3.5 would like to have asyncio as an option and also be able to use regular threads (for simple dev / testing); I was thinking something like what you did in graphene / graphql-core
• Add a simple pubsub class for testing / development; so the RedisPubsub doesn't have to be used
• Lot's of other things could be done obviously...my time being the limiting factor there. Once I finish the tests...obviously easier for folks to contribute.

Thanks!

[Helw150]

@hballard @syrusakbary With the official spec now merged, is there room for contributions on getting subscriptions implemented? I'm interested in using graphene with a new django project, and this would be huge.

[dudanogueira]

Indeed. This would make Django a great fit for angular2 real time apps

[hballard]

@Helw150 - I can't speak for @syrusakbary, but I know I'd welcome any contributions on my repo. Also, not sure if @syrusakbary is interested in integrating it into graphene eventually, just prefers to fork it, or go his own way. I started this mainly as a hobby project, when I was playing w/ Apollo subscriptions and noticed their wasn't an implementation for graphene (python being my preferred server language). I just pushed a commit to the "tests" branch with about half the subscriptions-transport tests and all the subscriptions-manager tests were added a few weeks ago. The rest of the transport tests should be easier to finish up now, the initial test setup for those slowed me down a bit--some of it was new for me. I haven't had a ton of time to devote to this and I don't really use django...so not sure when I would get to that. My next focus would be Python 3 compatibility...which might not be that difficult. Of course now that the initial commits for graphql subscriptions have been added to the spec, probably lot's more need to be done outside of these focues. You can read my summary of the my transport tests commit on the issue I created for them here

[hballard]

Update: Initial tests have been added as of last weekend (see commit here and I merged a commit last week that added Python 3 compatibility (2.7, 3.4, 3.5, & 3.6). My next two priorities are adding additional executors (threads, asyncio, etc.) and some type of Django compatibility. I don't really use Django...for those that do...is channels the preferred method for adding real-time services to Django now or something like django-websocket-redis or django-socketio? I've been doing a little reading on Django and channels...

[leebenson]

Great work @hballard. I'd like to use this as inspiration for an example in my ReactQL starter kit to show how subscriptions can be served from a non-Node.js server.

[hballard]

@leebenson - Very cool. Any feedback you can provide is appreciated. Let me know if I can be of assistance.

Per @syrusakbary previous comment above, "Having a subscriptions integration with Django so we assure that the subscriptions structure is abstracted in a scalable way..."; I've been thinking about the best way to do that.

My thought is that it should be fairly straightforward to generalize the concurrency executor methods like @syrusakbary did in graphql-core and have separate executor classes for each concurrency library (probably will even borrow some of the logic in graphql-core for each one). Then the RedisPubsub and SubscriptionTransport classes would utilize the corresponding executor passed in by the user when they instantiate each class. I'd welcome any feedback anyone has (@syrusakbary or others) on this structure.

I spent a couple hours reading through the Django channels library this weekend and it would seem it could be treated as just another executor class under this model. Also, anyone familiar w/ Django...seems like I would utilize a "class-based consumer" (inherit from WebsocketConsumer) for the SubscriptionServer. The redis pubsub "wait_and_get_message" method in the RedisPubsub class could be implemented as another consumer. Thoughts on this (from anyone more familiar w/ Django channels)?

[Eraldo]

How about getting someone from the django core team to assist?
I think reactive programming is something that would bring new people to django which would benefit in the long run. It was the reason for me to look into other solutions.
What do you think?

[hballard]

I'm happy to have any assistance from another contributor--particularly a django core. I haven't reached out to them since I don't utilize django and I needed to abstract the concurrency executor from the base subscriptions manager and websocket server logic, in order to use other concurrency frameworks (like django channels). It would be fairly straightforward to integrate the current gevent version with django-socketio or create a simple package similar to flask-sockets to integrate geventwebsocket into django directly. I found a small library on bitbucket that seems to do just that -- django-gevent-websocket (here).

I'm currently working on abstracting the currency executor to be able to use asyncio for concurrency as well (vs the current gevent). I should merge a version of that with master in the next week or so (which will allow use w/ Sanic web framework using uvloop / asyncio)...and then I was going to turn my attention to django integration, using the same abstraction base. But my plan was to focus more on django-channels...since that seems to be the way forward for django concurrency.

[Eraldo]

Thank you for the update @hballard that sounds really awesome. :D

[Eraldo]

I am using django in some projects and would love to replace drf (django-rest-framework) with what you are building. I really prefer reactive programming and I think a graphQL api with subscriptions would totally add a lot of value on top of the django project.
I am not a core django person nor do I have experience with django-channels.. I do however have practical experience with django. Feel free to reach out me if it helps. ;)

[patrick91]

@hballard will you be at EuroPython? maybe we can find someone to help during the sprints! :)

[hballard]

Afraid not...I live in Texas (US)...and that would be a bit of hike!

[Eraldo]

Did you get to give it a try yet? :)

[syrusakbary]

@Eraldo @hballard Yes, I think I should post an update here as I'm working full on subscriptions now.

Some thoughts about my journey: the way Apollo-Subscriptions use to manage subscriptions was not very friendly for the developer, needing to hack around the resolution and a specific PubSub implementation that was "bypassing" the GraphQL engine for adapting it into subscriptions.
The reason for that is the GraphQL-js engine was not ready for subscriptions (meaning that was only able to return either a promise or a static value, but not a async iterator).

However GraphQL-js recently added a way to subscribing to a GraphQL query (that return an async iterator a.k.a. Observable) that pushed towards simpler and cleaner implementations of subscriptions that decouple the subscription resolution from the "listener" on the subscription.

That led to better implementations of the transport mechanisms in GraphQL subscriptions like subscriptions-transport-ws.

So, in summary, subscriptions is something that should be bundled fully into the GraphQL engine, in a way that is easy to plug any mechanisms, such as:

• websockets with asyncio
• websockets with gevent
• websockets with django-daphne

That don't require any specific pub/sub implementation and, eventually, let this decision to the developer (in case it want to use it).

For the next version of Graphene, 2.0 I plan to have subscriptions bundled into the Engine :)
There is already a branch in graphql-core where I'm doing the research process.

I will keep updating this thread with more information as I keep working on it.

[ProjectCheshire]

@syrusakbary what ended being the recommended approach on this? I'm using graphql python and loving it, and I'm working on an iot device project where they use mqtt for data observations. Curious if there is a recommended approach or if it is 'do what thou wilt' for subscriptions in graphql python.

Thanks!

[AgentChris]

can we have an example of how subscriptions works with Graphene in Django?

[japrogramer]

@AgentChris
Here is how i managed it #500 (comment)
also look at graphql-python/graphql-core#149

[AgentChris]

thanks a los, i was trying to find an example for the last 2 days

[eamigo86]

Hi @AgentChris, take a look at this module, maybe you might be interested:
graphene-django-subscriptions

[Oxyrus]

I would like to know what is the progress in the implementation of subscriptions into the core of Graphene, could someone clarify please?

[japrogramer]

@Oxyrus it seems that the mechanism has been decided to be rx observables but the method for delivery to the client is still up to you.

[tricoder42]

Here's a gist with my solution using django-channels. It's a working proof of concept. Next task is optimize it for production environment. Feedback welcome!

[Musbell]

Ahhh! i have been waiting for this. @prokher i will definitely check it out.

[prokher]

@Musbell be careful, it is not ready for production yet, currently we are working hard to improve (add asynchronous message handling, ordering/serialization, examples, etc), and we are happy to receive any feedback you have.

[Musbell]

@prokher thanks for the notice. 👍

[blackflame007]

Subscriptions should be in Graphene2.0+ @syrusakbary mentioned it in this tweet https://twitter.com/syrusakbary/status/923325568157859840?lang=en
I can't find the function in the repo or any documentation, let's hope we get an update soon

[mvanlonden]

Subscriptions introduced in v2! https://github.com/graphql-python/graphene/releases/tag/v2.0.0

[japrogramer]

@mvanlonden correct me if im wrong but tag v2.0.0 has been out for a while ..
And i can see no commit mentioning subscriptions. in commits or in the tags that follow.

Could you provide an example implementation of a subscription.
would one need to use channels in django for this .. ?

[mvanlonden]

@japrogramer it is not documented well/at all but documentation needs are tracked here. This commit added support for subscriptions.

https://github.com/eamigo86/graphene-django-subscriptions is an example of using graphene-django with subscriptions although we'd like to implement the functionality natively in graphene-django

Mind submitting a PR with documentation for subscriptions?

[japrogramer]

@mvanlonden that has been possible for a while, i thought your announcement was about making the message delivered via graphene.

Not sure this issue should be closed, since the apollo integration still takes some set up.
even with a simple lib.

case and point.

at any rate, here is another successful attempt,
This time i wrote it for aiohttp

from graphql import GraphQLError
from aiograph.redistools.pubsub import subscribe, reader
from aiograph.redistools.utils import create_redis

from graphene.types import generic
import graphene
import asyncio


class TestType(graphene.ObjectType):
    name = "test"
    msg = graphene.String()


class Query(object):
    test_field = graphene.Field(TestType, msg=graphene.String())

    async def resolve_test_field(self, info, **args):
        msg = args.get('msg', None)
        return TestType(msg=msg)


class MessageMutation(graphene.Mutation):
    status = graphene.Int()
    errors = generic.GenericScalar()

    class Arguments:
        channel = graphene.String(required=True)
        msg = graphene.String(required=True)

    @staticmethod
    async def mutate(root, info, **input):
        try:
            redis = await create_redis()
            await redis.publish(
                input.get('channel'),
                input.get('msg'))
            return MessageMutation(status=200, errors=None)
        except Exception as e:
            breakpoint()
            raise GraphQLError('An Error occoured.')

class TestSubscription(object):
    test_updated = graphene.Field(TestType, channel=graphene.String())

    async def resolve_test_updated(root, info, **input):
        loop = asyncio.get_event_loop()
        def next(message):
            return TestType(msg=message)
        redis = await create_redis()
        channel = await subscribe(redis, input.get('channel', 'default'))
        async for i in reader(channel[0]):
            # Here we yield the actual object for the TestType
            yield next(i)

class Mutations(object):
    send_message = MessageMutation.Field()

from contextlib import suppress
from aiohttp_session import get_session
from aiohttp import web, WSMsgType

from graphql.execution.executors.asyncio import AsyncioExecutor
from graphql import graphql

from config.settings import settings
from config.schema import schema

import asyncio
import functools
import json
import rx

from .utils import format_response, run_aiograph

# import the logging library
import logging
logger = logging.getLogger('asyncio')

async def graph_handler(request):
    session = await get_session(request)
    last_visit = session['last_visit'] if 'last_visit' in session else None
    if request.content_type in ('application/graphql', 'application/json'):
        payload = json.loads(await request.text())


    with suppress(NameError):
        message = dict()
        message['request'] = request
        result = await run_aiograph(
            payload,
            executor=AsyncioExecutor(loop=asyncio.get_event_loop()),
            allow_subscriptions=False,
            return_promise=True,
            **{'context_value': message}
            )
        response = format_response(result)
        return web.json_response(response)


class WebSocket(web.View):
    def __init__(self, *args, **kwargs):
        self.loop = asyncio.get_event_loop()
        return super().__init__(*args, **kwargs)

    async def get(self):
        ws = web.WebSocketResponse()
        await ws.prepare(self.request)

        session = await get_session(self.request)
        if 'WebSockets' in self.request.app:
            self.request.app['websockets'].append(ws)
        else:
            self.request.app['websockets'] = ws

        async for msg in ws:
            if msg.type == WSMsgType.text:
                data = json.loads(msg.data)
                if data['type'] == 'close':
                    await ws.close()
                else:
                    if data['type'] == 'connection_init':
                        logger.debug('connection_init')
                        await ws.send_json(
                            { 'type': 'websocket.send',
                              'text': json.dumps({'type': 'connection_ack'})}
                            )
                    if data['type'] == 'start':
                        await self._start(ws, data)
                    if data['type'] == 'stop':
                        await ws.send_str(str(data))
            elif msg.type == WSMsgType.error:
                log.debug(
                    'ws connection closed with exception %s'\
                    % ws.exception())

        if 'WebSockets' in self.request.app:
            self.request.app['websockets'].remove(ws)
        logger.debug('websocket connection closed')
        return ws

    async def _start(self, ws, data):
            message = dict()
            message['request'] = self.request
            result = await run_aiograph(
                data,
                allow_subscriptions=True,
                return_promise=True,
                **{'context_value': message}
                )
            if isinstance(result, rx.Observable):
                logger.debug('result is an Observable')
                result.subscribe(lambda t: self.loop.create_task(self._send_result(ws, t)))
            else:
                response = format_response(result)
                await ws.send_json(response)

    async def _send_result(self, ws, result):
        await ws.send_json(format_response(result))

Since this was a quick test this sufficed.

<!doctype html>

<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Humble begginings</title>
  <meta name="description" content="Test bed">
  <meta name="author" content="me">
</head>

<body>
  <script language="JavaScript">
    socket = new WebSocket("ws://" + window.location.host + "/wsgql");
    socket.onmessage = function(e) {
      alert(JSON.parse(e.data));
    }
    socket.onopen = function(){
      alert('Connection to server started')
    }
    socket.onclose = function(event){
      if(event.wasClean){
        alert('Clean connection end')
      }else{
        alert('Connection broken')
      }
    };
    socket.onerror = function(error){
      alert(error);
    }
    socket.onopen = function(){
      alert('Connection to server started')
      var msg = {
        type: 'start',
        query:
        'subscription { testUpdated(channel:"test"){ msg }}',
      }
      socket.send(JSON.stringify(msg));
    }
    console.log('about to send message.')
  </script>
</body>
</html>

[ambientlight]

@japrogramer: wondering if it makes sense ditching graphene subscriptions altogether for the sake of apollo client integration if there is no out-of-the box support for it.

Say, most minimal setup: Django users running on top of Postgres could simply run a standalone subscriptions-transport-ws with graphql-postgres-subscriptions, in graphene endpoint mutations we just directly use postgres's pubsub with NOTIFY pinAdded; '{"pinAdded":{"title":"wow",", "id":0}}'

[joshourisman]

Personally, I wouldn't consider graphql-postgres-subscriptions a suitable replacement for subscription functionality in Graphene. Obviously different people will have different needs, but if I'm building a GraphQL API for a Python project, divorcing my subscriptions backend from the main app introduces some undesirable complications. Unless you're just providing full, open access to your database, it would mean having to re-implement whatever authentication and permissions are involved in accessing your database. If you're using Django, for example, you no longer have access to the authentication backend and would need to re-implement it in Node, at which point why are you bothering with Python/Django/Graphene at all?

[ambientlight]

@joshourisman: this makes a good point, thanks. (I just had a thought of passing temporary subscription tokens to clients on login for things they can subscribe to, then verifying them upon subscription on subscriptions-transport-ws side)

Basically, I have a legacy (somehow large) django app. I'm investigating into graphene integration on top of that and hooking up apollo client with angular / native iOS / native Android clients with hopes of reducing amount of logic needed to be written to support the apis. And this seemingly build-in subscription thing into Apollo clients across each platform (haven't used any graphql before today pretty much) looks promising for the first glance.

[japrogramer]

@ambientlight you can implement subscriptions with django-channels and apollo, ive done it, its not trivial. Here is an example implementation that is similar to mine, in the way it delivers the message to the client. https://github.com/datadvance/DjangoChannelsGraphqlWs

[demeralde]

Does anyone know when subscriptions will be implemented natively in graphene?

[japrogramer]

@dspacejs because graphene is very separated from the common python frameworks .. it is difficult to implement a delivery system for all of them in this package.

However i would like to see packages like graphene-django and graphene-flask to implement their own handler/view that works for their platform .. following the language API
closely enough so that it works with apollo. That way we don't have users trying to roll their own implementation every time someone wants subscriptions And all the eyes of the community are in one implementation for the best support.

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[prokher]

I didn't get, is there a native subscription support in Graphene or not. Shall I drop my https://github.com/datadvance/DjangoChannelsGraphqlWs or not yet?

[tomjohnhall]

@prokher please don't drop the only one I could get working 😄

[erickwilder]

Is there any updates on this subject?

[jaydenwindle]

Hi everyone! I just released a GraphQL subscriptions implementation for Graphene + Django based on an internal implementation we’ve been using in production for the last 6 months at Jetpack (https://tryjetpack.com/). It builds on @tricoder42's gist implementation and takes inspiration from the great work that @japrogramer, @eamigo86 and @prokher have done. Since this is the canonical Graphene subscriptions issue I thought I would post it here.

https://github.com/jaydenwindle/graphene-subscriptions

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[Eraldo]

Any updates? (Trying to keep the discussion alive.) :)

[adiberk]

I know we are on graphene3 at this point and this discussion is extremely old, but is there any definitive documentation on how to run subscriptions with graphene2 and flask? Tbh, even with graphene3 I don't see much documentation on integrating with flask. But regardless, I am stuck at 2 for now and require setting up subscriptions :(