jetblack-graphene 3.0b0

GraphQL Framework for Python

This is a forked version of graphene with support for subscriptions.

For executing a subscription, you can directly call the subscribe method on it. This method is async and must be awaited.

import asyncio
from datetime import datetime
from graphene import ObjectType, String, Schema, Field

# All schema require a query.
class Query(ObjectType):
    hello = String()

    def resolve_hello(root, info):
        return 'Hello, world!'

class Subscription(ObjectType):
    time_of_day = Field(String)

    async def subscribe_time_of_day(root, info):
        while True:
            yield { 'time_of_day': datetime.now().isoformat()}
            await asyncio.sleep(1)

SCHEMA = Schema(query=Query, subscription=Subscription)

async def main(schema):

    subscription = 'subscription { timeOfDay }'
    result = await schema.subscribe(subscription)
    async for item in result:
        print(item.data['timeOfDay'])

asyncio.run(main(SCHEMA))

The result is an async iterator which yields items in the same manner as a query.