Graphene 3 example subscription

graphene3-demo-subscription.py

import typing

import graphene
from jetblack_tweeter import Tweeter

from .types import TweetType

class Subscription(graphene.ObjectType):
    filter = graphene.Field(
        TweetType,
        track=graphene.List(graphene.String, required=True))

    @staticmethod
    async def subscribe_filter(root, info, track: typing.List[str]):
        tweeter: Tweeter = info.context['tweeter']
        async for tweet in tweeter.stream.filter(track=track):
            yield tweet