test_subscription_transport.py 文件源码

def schema(data):
    class UserType(graphene.ObjectType):
        id = graphene.String()
        name = graphene.String()

    class Query(graphene.ObjectType):
        test_string = graphene.String()

    class Subscription(graphene.ObjectType):
        user = graphene.Field(UserType, id=graphene.String())
        user_filtered = graphene.Field(UserType, id=graphene.String())
        context = graphene.String()
        error = graphene.String()

        def resolve_user(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_user_filtered(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_context(self, args, context, info):
            return context

        def resolve_error(self, args, context, info):
            raise Exception('E1')

    return graphene.Schema(query=Query, subscription=Subscription)