Added support for subscription

docs/execution/execute.rst

@@ -17,6 +17,43 @@ For executing a query a schema, you can directly call the ``execute`` method on
 ``result`` represents the result of execution. ``result.data`` is the result of executing the query, ``result.errors`` is ``None`` if no errors occurred, and is a non-empty list if an error occurred.
 
 
+For executing a subscription, you can directly call the ``subscribe`` method on it.
+This method is async and must be awaited.
+
+.. code:: python
+
+    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.
+
 .. _SchemaExecuteContext:
 
 Context

graphene/types/schema.py

@@ -8,7 +8,9 @@ from graphql import (
     graphql_sync,
     introspection_types,
     is_type,
+    parse,
     print_schema,
+    subscribe,
     GraphQLArgument,
     GraphQLBoolean,
     GraphQLEnumValue,
@@ -365,7 +367,10 @@ class GrapheneGraphQLSchema(GraphQLSchema):
                     field_type,
                     args=args,
                     resolve=field.get_resolver(
-                        self.get_resolver_for_type(type_, name, field.default_value)
+                        self.get_resolver_for_type(type_, "resolve_{}", name, field.default_value)
+                    ),
+                    subscribe=field.get_resolver(
+                        self.get_resolver_for_type(type_, "subscribe_{}", name, field.default_value)
                     ),
                     deprecation_reason=field.deprecation_reason,
                     description=field.description,
@@ -374,10 +379,11 @@ class GrapheneGraphQLSchema(GraphQLSchema):
             fields[field_name] = _field
         return fields
 
-    def get_resolver_for_type(self, type_, name, default_value):
+    def get_resolver_for_type(self, type_, pattern, name, default_value):
         if not issubclass(type_, ObjectType):
             return
-        resolver = getattr(type_, "resolve_{}".format(name), None)
+        func_name = pattern.format(name)
+        resolver = getattr(type_, func_name, None)
         if not resolver:
             # If we don't find the resolver in the ObjectType class, then try to
             # find it in each of the interfaces
@@ -385,7 +391,7 @@ class GrapheneGraphQLSchema(GraphQLSchema):
             for interface in type_._meta.interfaces:
                 if name not in interface._meta.fields:
                     continue
-                interface_resolver = getattr(interface, "resolve_{}".format(name), None)
+                interface_resolver = getattr(interface, func_name, None)
                 if interface_resolver:
                     break
             resolver = interface_resolver
@@ -505,6 +511,11 @@ class Schema:
         kwargs = normalize_execute_kwargs(kwargs)
         return await graphql(self.graphql_schema, *args, **kwargs)
 
+    async def subscribe(self, query, *args, **kwargs):
+        document = parse(query)
+        kwargs = normalize_execute_kwargs(kwargs)
+        return await subscribe(self.graphql_schema, document, *args, **kwargs)
+
     def introspect(self):
         introspection = self.execute(introspection_query)
         if introspection.errors:

tests_asyncio/test_subscribe.py

@@ -0,0 +1,30 @@
+from pytest import mark
+
+from datetime import datetime
+from graphene import ObjectType, Int, String, Schema, Field
+
+class Query(ObjectType):
+    hello = String()
+
+    def resolve_hello(root, info):
+        return 'Hello, world!'
+
+class Subscription(ObjectType):
+    count_to_ten = Field(Int)
+
+    async def subscribe_count_to_ten(root, info):
+        count = 0
+        while count < 10:
+            count += 1
+            yield { 'count_to_ten': count }
+
+schema = Schema(query=Query, subscription=Subscription)
+
+@mark.asyncio
+async def test_subscription():
+    subscription = 'subscription { countToTen }'
+    result = await schema.subscribe(subscription)
+    count = 0
+    async for item in result:
+        count = item.data['countToTen']
+    assert count == 10