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

def gen_operators_of_field(f_name, mongo_field, r_graphene, operators_list):
    """ Return a dict with keys as the name of the field with operator and value is the required type, for instance: 
    @param f_name: string name of the field
    @param mongo_field: object instance of mongoengine field, e.g: mongoengine.StringField()
    @param r_graphene: object instance of graphene field, e.g: graphene.String(): 
    {
        name: graphene.String()
        name__nin: graphene.List(graphene.String) ...
    }
     """

    field_with_operators = {
        f_name: field_to_id(mongo_field, r_graphene)
    }

    for op_name in operators_list:
        field_with_operators[f_name + '__' + op_name] = operators[op_name](mongo_field, r_graphene)

    if isinstance(mongo_field, fields_string_operators):
        for op in string_operators:
            field_with_operators[f_name + '__' + op] = graphene.String()

    return field_with_operators

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_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 validation_schema():
    class Query(graphene.ObjectType):
        placeholder = graphene.String()

    class Subscription(graphene.ObjectType):
        test_1 = graphene.String()
        test_2 = graphene.String()

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

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)

def test_should_call_unsubscribe_when_client_closes_cxn(server_with_mocks):
    node_script = '''
        module.paths.push('{0}')
        WebSocket = require('ws')
        const SubscriptionClient =
        require('subscriptions-transport-ws').SubscriptionClient
        const client = new SubscriptionClient('ws://localhost:{1}/socket')
        client.subscribe({{
            query: `subscription useInfo($id: String) {{
            user(id: $id) {{
              id
              name
            }}
          }}`,
            operationName: 'useInfo',
            variables: {{
              id: 3,
            }},
          }}, function (error, result) {{
            // nothing
          }}
        )
        setTimeout(() => {{
            client.client.close()
        }}, 500)
    '''.format(
        os.path.join(os.path.dirname(__file__), 'node_modules'), TEST_PORT)
    try:
        subprocess.check_output(
            ['node', '-e', node_script], stderr=subprocess.STDOUT, timeout=1)
    except:
        while True:
            mock = server_with_mocks.get_nowait()
            if mock.name == 'on_unsubscribe':
                mock.assert_called_once()
                break

def test_should_trigger_on_unsubscribe_when_client_unsubscribes(
        server_with_mocks):
    node_script = '''
        module.paths.push('{0}')
        WebSocket = require('ws')
        const SubscriptionClient =
        require('subscriptions-transport-ws').SubscriptionClient
        const client = new SubscriptionClient('ws://localhost:{1}/socket')
        const subId = client.subscribe({{
            query: `subscription useInfo($id: String) {{
            user(id: $id) {{
              id
              name
            }}
          }}`,
            operationName: 'useInfo',
            variables: {{
              id: 3,
            }},
          }}, function (error, result) {{
            // nothing
          }})
        client.unsubscribe(subId)
    '''.format(
        os.path.join(os.path.dirname(__file__), 'node_modules'), TEST_PORT)
    try:
        subprocess.check_output(
            ['node', '-e', node_script], stderr=subprocess.STDOUT, timeout=.2)
    except:
        while True:
            mock = server_with_mocks.get_nowait()
            if mock.name == 'on_unsubscribe':
                mock.assert_called_once()
                break

def convert_column_to_string(type, attribute, registry=None):
    if attribute.is_hash_key:
        return ID(description=attribute.attr_name, required=not attribute.null)

    return String(description=getattr(attribute, 'attr_name'),
                  required=not (getattr(attribute, 'null', True)))

def convert_date_to_string(type, attribute, registry=None):
    return String(description=getattr(attribute, 'attr_name'),
                  required=not (getattr(attribute, 'null', True)))

def convert_scalar_list_to_list(type, attribute, registry=None):
    return List(String, description=attribute.attr_name)

def convert_list_to_list(type, attribute, registry=None):
    return List(String, description=attribute.attr_name)

def test_should_datetime_convert_string():
    assert_attribute_conversion(UTCDateTimeAttribute(), graphene.String)

def test_should_binary_convert_string():
    assert_attribute_conversion(BinaryAttribute(), graphene.String)

def convert_form_field_to_string(field):
    return String(description=field.help_text, required=field.required)

def test_should_char_convert_string():
    assert_conversion(models.CharField, graphene.String)

def test_should_text_convert_string():
    assert_conversion(models.TextField, graphene.String)

def test_should_slug_convert_string():
    assert_conversion(models.SlugField, graphene.String)

def test_should_url_convert_string():
    assert_conversion(models.URLField, graphene.String)

def test_should_ipaddress_convert_string():
    assert_conversion(models.GenericIPAddressField, graphene.String)

def test_should_file_convert_string():
    assert_conversion(models.FileField, graphene.String)

def test_should_image_convert_string():
    assert_conversion(models.ImageField, graphene.String)

def test_should_time_convert_string():
    assert_conversion(forms.TimeField, graphene.String)

def test_should_date_time_convert_string():
    assert_conversion(forms.DateTimeField, graphene.String)

def test_should_char_convert_string():
    assert_conversion(forms.CharField, graphene.String)

def test_should_email_convert_string():
    assert_conversion(forms.EmailField, graphene.String)

def test_should_slug_convert_string():
    assert_conversion(forms.SlugField, graphene.String)

def test_should_choice_convert_string():
    assert_conversion(forms.ChoiceField, graphene.String)

def test_should_base_field_convert_string():
    assert_conversion(forms.Field, graphene.String)

def test_should_regex_convert_string():
    assert_conversion(forms.RegexField, graphene.String, '[0-9]+')

def convert_serializer_field_to_list_of_string(field):
    return (graphene.List, graphene.String)

def test_should_char_convert_string():
    assert_conversion(serializers.CharField, graphene.String)

def test_should_slug_convert_string():
    assert_conversion(serializers.SlugField, graphene.String)

def test_should_url_convert_string():
    assert_conversion(serializers.URLField, graphene.String)

def test_should_choice_convert_string():
    assert_conversion(serializers.ChoiceField, graphene.String, choices=[])

def test_should_base_field_convert_string():
    assert_conversion(serializers.Field, graphene.String)

def test_should_regex_convert_string():
    assert_conversion(serializers.RegexField, graphene.String, regex='[0-9]+')

def test_should_duration_convert_string():
    assert_conversion(serializers.DurationField, graphene.String)

def test_should_file_convert_string():
    assert_conversion(serializers.FileField, graphene.String)

def test_should_filepath_convert_string():
    assert_conversion(serializers.FilePathField, graphene.String, path='/')

def test_should_ip_convert_string():
    assert_conversion(serializers.IPAddressField, graphene.String)

def test_should_image_convert_string():
    assert_conversion(serializers.ImageField, graphene.String)

def convert_column_to_string(type, column, registry=None):
    return graphene.String(
        description=get_column_doc(column),
        required=not(is_column_nullable(column)),
    )

def convert_ndb_string_property(ndb_prop, registry=None):
    return convert_ndb_scalar_property(String, ndb_prop)

def convert_ndb_key_propety(ndb_key_prop, registry=None):
    """
    Two conventions for handling KeyProperties:
    #1.
        Given:
            store_key = ndb.KeyProperty(...)

        Result is 2 fields:
            store_id  = graphene.String() -> resolves to store_key.urlsafe()
            store     = NdbKeyField()     -> resolves to entity

    #2.
        Given:
            store = ndb.KeyProperty(...)

        Result is 2 fields:
            store_id = graphene.String() -> resolves to store_key.urlsafe()
            store     = NdbKeyField()    -> resolves to entity

    """
    is_repeated = ndb_key_prop._repeated
    name = ndb_key_prop._code_name

    if name.endswith('_key') or name.endswith('_keys'):
        # Case #1 - name is of form 'store_key' or 'store_keys'
        string_prop_name = rreplace(name, '_key', '_id', 1)
        resolved_prop_name = name[:-4] if name.endswith('_key') else p.plural(name[:-5])
    else:
        # Case #2 - name is of form 'store'
        singular_name = p.singular_noun(name) if p.singular_noun(name) else name
        string_prop_name = singular_name + '_ids' if is_repeated else singular_name + '_id'
        resolved_prop_name = name

    return [
        ConversionResult(name=string_prop_name, field=DynamicNdbKeyStringField(ndb_key_prop, registry=registry)),
        ConversionResult(name=resolved_prop_name, field=DynamicNdbKeyReferenceField(ndb_key_prop, registry=registry))
    ]

def convert_computed_property(ndb_computed_prop, registry=None):
    return convert_ndb_scalar_property(String, ndb_computed_prop)

def testGET_support_json_variables(self):
        response = self.app.get('/graphql', params=dict(
            query='query helloWho($who: String){ greet(who: $who) }',
            variables=json.dumps({'who': "ekampf"})
        ))

        response_dict = json.loads(response.body)
        self.assertDictEqual(
            response_dict.get('data'), {'greet': 'Hello ekampf!'}
        )

def testPOST_support_json_variables(self):
        response = self.app.post('/graphql', params=json.dumps(dict(
            query='query helloWho($who: String){ greet(who: $who) }',
            variables={'who': "ekampf"}
        )))

        response_dict = json.loads(response.body)
        self.assertDictEqual(
            response_dict.get('data'), {'greet': 'Hello ekampf!'}
        )

def test_handles_poorly_formed_variables(self):
        for method in (self.get, self.post):
            response = method('/graphql', expect_errors=True, params=dict(
                query='query helloWho($who: String){ greet(who: $who) }',
                variables='who:You'
            ))

            response_data = json.loads(response.body)
            self.assertEqual(response.status_int, 400)
            self.assertEqual(response_data['errors'][0]['message'], 'Variables are invalid JSON.')

def testStringProperty_shouldConvertToString(self):
        self.__assert_conversion(ndb.StringProperty, graphene.String)