Python graphene 模块，Boolean() 实例源码

def __init__(self, ndb_key_prop, graphql_type_name, *args, **kwargs):
        self.__ndb_key_prop = ndb_key_prop
        self.__graphql_type_name = graphql_type_name
        is_repeated = ndb_key_prop._repeated
        is_required = ndb_key_prop._required

        _type = String
        if is_repeated:
            _type = List(_type)

        if is_required:
            _type = NonNull(_type)

        kwargs['args'] = {
            'ndb': Argument(Boolean, False, description="Return an NDB id (key.id()) instead of a GraphQL global id")
        }

        super(NdbKeyStringField, self).__init__(_type, *args, **kwargs)

def get_field_args(field):
    types = {
        str: graphene.String,
        bool: graphene.Boolean,
        int: graphene.Int,
        float: graphene.Float,
    }

    try:
        args = inspect.signature(field)
        result = {}
        for name, arg in args.parameters.items():
            if arg.default is not None:
                custom_type = types[type(arg.default)]
            else:
                custom_type = graphene.String

            result.update({name: custom_type()})
        return result
    except (KeyError, ValueError):
        return {}

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

        def resolve_test_string(self, args, context, info):
            return 'works'

    # TODO: Implement case conversion for arg names
    class Subscription(graphene.ObjectType):
        test_subscription = graphene.String()
        test_context = graphene.String()
        test_filter = graphene.String(filterBoolean=graphene.Boolean())
        test_filter_multi = graphene.String(
            filterBoolean=graphene.Boolean(),
            a=graphene.String(),
            b=graphene.Int())
        test_channel_options = graphene.String()

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

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

        def resolve_test_filter(self, args, context, info):
            return 'good_filter' if args.get('filterBoolean') else 'bad_filter'

        def resolve_test_filter_multi(self, args, context, info):
            return 'good_filter' if args.get('filterBoolean') else 'bad_filter'

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

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

def test_use_filter_functions_properly(sub_mgr):
    query = 'subscription Filter1($filterBoolean: Boolean) {\
    testFilter(filterBoolean: $filterBoolean)}'

    def callback(err, payload):
        if err:
            sys.exit(err)
        else:
            try:
                if payload is None:
                    assert True
                else:
                    assert payload.data.get('testFilter') == 'good_filter'
                    sub_mgr.pubsub.greenlet.kill()
            except AssertionError as e:
                sys.exit(e)

    def publish_and_unsubscribe_handler(sub_id):
        sub_mgr.publish('filter_1', {'filterBoolean': False})
        sub_mgr.publish('filter_1', {'filterBoolean': True})
        sub_mgr.pubsub.greenlet.join()
        sub_mgr.unsubscribe(sub_id)

    p1 = sub_mgr.subscribe(query, 'Filter1', callback, {'filterBoolean': True},
                           {}, None, None)
    p2 = p1.then(publish_and_unsubscribe_handler)
    p2.get()

def test_use_filter_func_that_returns_a_promise(sub_mgr):
    query = 'subscription Filter2($filterBoolean: Boolean) {\
    testFilter(filterBoolean: $filterBoolean)}'

    def callback(err, payload):
        if err:
            sys.exit(err)
        else:
            try:
                if payload is None:
                    assert True
                else:
                    assert payload.data.get('testFilter') == 'good_filter'
                    sub_mgr.pubsub.greenlet.kill()
            except AssertionError as e:
                sys.exit(e)

    def publish_and_unsubscribe_handler(sub_id):
        sub_mgr.publish('filter_2', {'filterBoolean': False})
        sub_mgr.publish('filter_2', {'filterBoolean': True})
        try:
            sub_mgr.pubsub.greenlet.join()
        except:
            raise
        sub_mgr.unsubscribe(sub_id)

    p1 = sub_mgr.subscribe(query, 'Filter2', callback, {'filterBoolean': True},
                           {}, None, None)
    p2 = p1.then(publish_and_unsubscribe_handler)
    p2.get()

def test_can_subscribe_to_more_than_one_trigger(sub_mgr):
    non_local = {'trigger_count': 0}

    query = 'subscription multiTrigger($filterBoolean: Boolean,\
            $uga: String){testFilterMulti(filterBoolean: $filterBoolean,\
            a: $uga, b: 66)}'

    def callback(err, payload):
        if err:
            sys.exit(err)
        else:
            try:
                if payload is None:
                    assert True
                else:
                    assert payload.data.get('testFilterMulti') == 'good_filter'
                    non_local['trigger_count'] += 1
            except AssertionError as e:
                sys.exit(e)
        if non_local['trigger_count'] == 2:
            sub_mgr.pubsub.greenlet.kill()

    def publish_and_unsubscribe_handler(sub_id):
        sub_mgr.publish('not_a_trigger', {'filterBoolean': False})
        sub_mgr.publish('trigger_1', {'filterBoolean': True})
        sub_mgr.publish('trigger_2', {'filterBoolean': True})
        sub_mgr.pubsub.greenlet.join()
        sub_mgr.unsubscribe(sub_id)

    p1 = sub_mgr.subscribe(query, 'multiTrigger', callback,
                           {'filterBoolean': True,
                            'uga': 'UGA'}, {}, None, None)
    p2 = p1.then(publish_and_unsubscribe_handler)
    p2.get()

def test_calls_the_error_callback_if_there_is_an_execution_error(sub_mgr):
    query = 'subscription X($uga: Boolean!){\
        testSubscription  @skip(if: $uga)\
    }'

    def callback(err, payload):
        try:
            assert payload is None
            assert err.message == 'Variable "$uga" of required type\
 "Boolean!" was not provided.'

            sub_mgr.pubsub.greenlet.kill()
        except AssertionError as e:
            sys.exit(e)

    def unsubscribe_and_publish_handler(sub_id):
        sub_mgr.publish('testSubscription', 'good')
        try:
            sub_mgr.pubsub.greenlet.join()
        except AttributeError:
            return
        sub_mgr.unsubscribe(sub_id)

    p1 = sub_mgr.subscribe(
        query,
        operation_name='X',
        callback=callback,
        variables={},
        context={},
        format_error=None,
        format_response=None)
    p2 = p1.then(unsubscribe_and_publish_handler)
    p2.get()

def convert_column_to_boolean(type, attribute, registry=None):
    return Boolean(description=attribute.attr_name, required=not attribute.null)

def test_should_boolean_convert_boolean():
    assert_attribute_conversion(BooleanAttribute(), graphene.Boolean)

def convert_form_field_to_boolean(field):
    return Boolean(description=field.help_text, required=True)

def convert_form_field_to_nullboolean(field):
    return Boolean(description=field.help_text)

def test_should_nullboolean_convert_boolean():
    assert_conversion(models.NullBooleanField, graphene.Boolean)

def test_should_boolean_convert_boolean():
    field = assert_conversion(forms.BooleanField, graphene.Boolean)
    assert isinstance(field.type, NonNull)

def test_should_nullboolean_convert_boolean():
    field = assert_conversion(forms.NullBooleanField, graphene.Boolean)
    assert not isinstance(field.type, NonNull)

def plural():
        return graphene.List(StatementGraph, is_startpoint=graphene.Boolean(), issue_uid=graphene.Int())

def __init__(self, type, transform_edges=None, *args, **kwargs):
        super(NdbConnectionField, self).__init__(
            type,
            *args,
            keys_only=Boolean(),
            batch_size=Int(),
            page_size=Int(),
            **kwargs
        )

        self.transform_edges = transform_edges

def convert_ndb_boolean_property(ndb_prop, registry=None):
    return convert_ndb_scalar_property(Boolean, ndb_prop)

def testBoolProperty_shouldConvertToString(self):
        self.__assert_conversion(ndb.BooleanProperty, graphene.Boolean)

def gen_mutation(model, graphene_schema, operators_mutation, fields_mutation, mutate_func, validator):
    """ We need to create a class that seems as follows (http://docs.graphene-python.org/en/latest/types/mutations/):

    class CreatePerson(graphene.Mutation):
        class Input:
            name = graphene.String()

        ok = graphene.Boolean()
        person = graphene.Field(lambda: Person)

        @staticmethod
        def mutate(root, args, context, info):
            person = Person(name=args.get('name'))
            ok = True
            return CreatePerson(person=person, ok=ok) 
    """

    def user_mutate(root, args, context, info):
        if validator:
            validator(model, args, {}, {})

        obj = mutate_func(args, context)
        if not isinstance(obj, model):
            raise TypeError('Failed to resolve mutation of the schema {}'
                            ' because mutate function must return a instance of {}, and the return type was {}.'
                            .format(graphene_schema.__name__, model.__name__, type(obj)))

        graphene_obj = mongo_to_graphene(obj, graphene_schema, fields_mutation)
        return Create(**{to_snake_case(model.__name__): graphene_obj})

    def generic_mutate(root, args, context, info):
        if validator:
            validator(model, args, {}, {})

        obj = model(**args)
        obj.save()
        graphene_obj = mongo_to_graphene(obj, graphene_schema, fields_mutation)
        return Create(**{to_snake_case(model.__name__): graphene_obj})

    Create = type('Create' + model.__name__, (graphene.Mutation,), {
        'Input': type('Input', (), operators_mutation),
        to_snake_case(model.__name__): graphene.Field(lambda: graphene_schema),
        'mutate': staticmethod(generic_mutate) if not mutate_func else staticmethod(user_mutate)
    })

    return Create