Skip to content

Instantly share code, notes, and snippets.

@dfee
Created November 25, 2017 03:17
Show Gist options
  • Save dfee/ca0aaff06d250b10c8f39fb4f862517a to your computer and use it in GitHub Desktop.
Save dfee/ca0aaff06d250b10c8f39fb4f862517a to your computer and use it in GitHub Desktop.
Example of how subscriptions work with graphene
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'
)
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment