Example of how subscriptions work with graphene

graphene_subscription.py

from collections import OrderedDict

import graphene
import rx


subject = rx.subjects.Subject()


class Author(graphene.ObjectType):
    registry = {}
    id = graphene.ID()
    name = graphene.String()

    @classmethod
    def create(cls, name):
        id_ = len(cls.registry)
        author = cls(id=id_, name=name)
        cls.registry[id_] = author
        return author


class CreateAuthor(graphene.Mutation):
    class Arguments:
        name = graphene.String(required=True)

    author = graphene.Field(Author)

    @staticmethod
    def mutate(root, info, **kwargs):
        # pylint: disable=W0613, unused-argument
        author = Author.create(**kwargs)
        return CreateAuthor(author)


class Query(graphene.ObjectType):
    author = graphene.Field(Author, id=graphene.ID(required=True))

    def resolve_author(self, info, id):
        # pylint: disable=R0201, no-self-use
        # pylint: disable=W0622, redefined-builtin
        # pylint: disable=W0613, unused-argument
        return Author.registry[int(id)]


class Mutation(graphene.ObjectType):
    create_author = CreateAuthor.Field()


class Subscription(graphene.ObjectType):
    authors = graphene.Field(Author, id=graphene.Int())

    def resolve_authors(self, info):
        # pylint: disable=W0613, unused-argument
        def _resolve(id):
            # pylint: disable=W0622, redefined-builtin
            return Author.registry[int(id)]

        return subject.\
            filter(lambda msg: msg[0] == 'authors').\
            map(lambda msg: _resolve(msg[1]))


schema = graphene.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription,
)

Author.registry[0] = Author(id=0, name='Bill')


def test_query_author():
    statement = '''
    query {
        author(id: 0) {
            id
            name
        }
    }'''
    result = schema.execute(statement)
    assert not result.errors
    assert result.data == OrderedDict(
        author=OrderedDict(
            id='0',
            name='Bill'
        ),
    )


def test_create_author():
    statement = '''
    mutation {
        createAuthor(name: "Joe") {
            author {
                id
                name
            }
        }
    }'''
    result = schema.execute(statement)
    assert not result.errors
    assert result.data == OrderedDict(
        createAuthor=OrderedDict(
            author=OrderedDict(
                id='1',
                name='Joe'
            ),
        ),
    )


def test_subscribe_author():
    statement = '''
    subscription {
        authors {
            id
            name
        }
    }
    '''
    observable = schema.execute(statement, allow_subscriptions=True)
    assert isinstance(observable, rx.Observable)

    results = []
    observable.subscribe(results.append)
    subject.on_next(('authors', 0))

    assert len(results) == 1
    result = results[0]
    assert not result.errors
    assert result.data == OrderedDict(
        authors=OrderedDict(
            id='0',
            name='Bill'
        )
    )