GraphQL Framework for Python
Project description
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.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
jetblack-graphene-3.0b0.tar.gz
(31.5 kB
view hashes)
Built Distribution
Close
Hashes for jetblack_graphene-3.0b0-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | ba96d7ce5855ac969a356a534c8898f1dddcaedda920ef381f1ac6d6899a6270 |
|
MD5 | 4ce267bc1fddb168a00c7da9132b3845 |
|
BLAKE2b-256 | fb74142edb442cb9b00fe4d7abe7ae55d1dfcccd0e39efde8969cab51565f8d8 |