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

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 list_field(f_name, mongo_field):
    from graphene_mongo import MongoSchema
    from graphene_mongo.fields.respective import respective_fields

    list_items_type = type(mongo_field.field)

    if list_items_type in respective_fields:
        return graphene.List(type(respective_fields[list_items_type]()))
    else:
        try:
            document = mongo_field.field.document_type
        except AttributeError:
            raise AttributeError('Error in {} field, have sure that this is defined with a mongoengine field'
                                 .format(f_name))

        schema = MongoSchema.get_or_generate_schema(document).schema
        return graphene.List(schema)

def test_add_operators_to_field_list_field():
    from mongoengine import ListField, SortedListField
    from graphene_mongo.operators import gen_operators_of_field, allowed_operators
    from graphene_mongo.fields.respective import respective_special_fields, respective_fields

    for m_field in [ListField, SortedListField]:
        for f, r_graphene in respective_fields.items():
            field = m_field(f())

            applied_operators = gen_operators_of_field('test', field, respective_special_fields[m_field],
                                                       allowed_operators(field))

            expected = format_fields(['size'])

            assert len(applied_operators.keys()) == len(expected)
            assert sorted(list(applied_operators.keys())) == sorted(expected)

            obj_list_field = applied_operators['test']('listTest', field)
            assert isinstance(obj_list_field, graphene.List)

            # here we test to assert that the type of items of the list is what is suppose to be
            assert isinstance(obj_list_field.of_type, type(r_graphene))

def test_should_manytoone_convert_connectionorlist():
    # Django 1.9 uses 'rel', <1.9 uses 'related
    related = getattr(Reporter.articles, 'rel', None) or \
        getattr(Reporter.articles, 'related')

    class A(DjangoObjectType):

        class Meta:
            model = Article

    graphene_field = convert_django_field(related, A._meta.registry)
    assert isinstance(graphene_field, graphene.Dynamic)
    dynamic_field = graphene_field.get_type()
    assert isinstance(dynamic_field, graphene.Field)
    assert isinstance(dynamic_field.type, graphene.List)
    assert dynamic_field.type.of_type == A

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 convert_local_structured_property(ndb_structured_property, registry=None):
    is_required = ndb_structured_property._required
    is_repeated = ndb_structured_property._repeated
    model = ndb_structured_property._modelclass
    name = ndb_structured_property._code_name

    def dynamic_type():
        _type = registry.get_type_for_model(model)
        if not _type:
            return None

        if is_repeated:
            _type = List(_type)

        if is_required:
            _type = NonNull(_type)

        return Field(_type)

    field = Dynamic(dynamic_type)
    return ConversionResult(name=name, field=field)

def testQuery_excludedField(self):
        Article(headline="h1", summary="s1").put()

        class ArticleType(NdbObjectType):
            class Meta:
                model = Article
                exclude_fields = ['summary']

        class QueryType(graphene.ObjectType):
            articles = graphene.List(ArticleType)

            def resolve_articles(self, info):
                return Article.query()

        schema = graphene.Schema(query=QueryType)
        query = '''
            query ArticlesQuery {
              articles { headline, summary }
            }
        '''

        result = schema.execute(query)

        self.assertIsNotNone(result.errors)
        self.assertTrue('Cannot query field "summary"' in result.errors[0].message)

def resolve_address(self, args, info):
        addresses = getattr(self, 'addresses', [])
        address = addresses[0] if addresses else {}
        addressStr = self.address_str(address)
        title = address.get('title', '')
        addressDetail = address.get('address', None)
        country = address.get('country', None)
        city = address.get('city', None)
        zipcode = address.get('zipcode', None)
        department = address.get('department', None)
        geoLocation = address.get('coordinates', None)

        return [Address(
            title=title,
            address=addressDetail,
            country=country,
            city=city,
            zipcode=zipcode,
            department=department,
            addressStr=addressStr,
            geoLocation=geoLocation)]


# class TimeInterval(graphene.ObjectType):
#     start = graphene.core.types.custom_scalars.DateTime()
#     end = graphene.core.types.custom_scalars.DateTime()
#
#
# class ScheduleDate(graphene.ObjectType):
#     date = graphene.core.types.custom_scalars.DateTime()
#     time_intervals = graphene.List(TimeInterval)

def convert_relationship_to_dynamic(type, attribute, registry=None):
    def dynamic_type():
        _type = registry.get_type_for_model(attribute.model)
        if not _type:
            return None

        if isinstance(attribute, OneToOne):
            return Field(_type)

        if isinstance(attribute, OneToMany):
            if is_node(_type):
                return PynamoConnectionField(_type)
            return Field(List(_type))

    return Dynamic(dynamic_type)

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_string_set_convert_list():
    assert_attribute_conversion(UnicodeSetAttribute(), graphene.List)

def test_should_number_set_convert_list():
    assert_attribute_conversion(NumberSetAttribute(), graphene.List)

def test_should_onetomany_convert_nonnode_field():
    class A(PynamoObjectType):
        class Meta:
            model = Article

    dynamic_field = convert_pynamo_attribute(Reporter.articles, Reporter.articles, A._meta.registry)
    assert isinstance(dynamic_field, Dynamic)
    graphene_type = dynamic_field.get_type()
    assert isinstance(graphene_type, graphene.Field)
    assert graphene_type.type == graphene.List(A)

def test_should_list_convert_list():
    assert_attribute_conversion(ListAttribute(), graphene.List)

def resolver(g_schema, mongo_doc, operators_single=None, operators_list=None, is_list=False, validator=None):
        def auto_resolver(root, args, contex, info):
            return resolver_query(g_schema, mongo_doc, args, info, is_list=is_list, validator=validator)

        if is_list:
            return graphene.List(g_schema, **operators_list, resolver=auto_resolver)
        else:
            return graphene.Field(g_schema, **operators_single, resolver=auto_resolver)

def field_to_id(m_field, g_field):
    """ We need this because if we want to do a query using the id, we will pass a string to args with the id of the 
    document that we want, but graphene needs a ID field, instead of Field. This function convert to right thing."""

    if isinstance(m_field, ReferenceField):
        return graphene.ID()
    elif (isinstance(m_field, ListField) or isinstance(m_field, SortedListField)) and \
            isinstance(m_field.field, ReferenceField):
        """ Pass here if it is a ListField or SortedListField of ReferenceField """
        return graphene.List(graphene.ID)
    else:
        return g_field

def convert_form_field_to_list(field):
    return List(ID, required=field.required)

def test_should_manytomany_convert_connectionorlist_list():
    class A(DjangoObjectType):

        class Meta:
            model = Reporter

    graphene_field = convert_django_field(Reporter._meta.local_many_to_many[0], A._meta.registry)
    assert isinstance(graphene_field, graphene.Dynamic)
    dynamic_field = graphene_field.get_type()
    assert isinstance(dynamic_field, graphene.Field)
    assert isinstance(dynamic_field.type, graphene.List)
    assert dynamic_field.type.of_type == A

def test_should_postgres_array_convert_list():
    field = assert_conversion(ArrayField, graphene.List, models.CharField(max_length=100))
    assert isinstance(field.type, graphene.NonNull)
    assert isinstance(field.type.of_type, graphene.List)
    assert field.type.of_type.of_type == graphene.String

def test_should_postgres_range_convert_list():
    from django.contrib.postgres.fields import IntegerRangeField
    field = assert_conversion(IntegerRangeField, graphene.List)
    assert isinstance(field.type, graphene.NonNull)
    assert isinstance(field.type.of_type, graphene.List)
    assert field.type.of_type.of_type == graphene.Int

def test_should_multiple_choice_convert_connectionorlist():
    field = forms.ModelMultipleChoiceField(Reporter.objects.all())
    graphene_type = convert_form_field(field)
    assert isinstance(graphene_type, List)
    assert graphene_type.of_type == ID

def convert_serializer_field_to_list(field, is_input=True):
    child_type = get_graphene_type_from_serializer_field(field.child)

    return (graphene.List, child_type)

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

def test_should_list_convert_to_list():
    class StringListField(serializers.ListField):
        child = serializers.CharField()

    field_a = assert_conversion(
        serializers.ListField,
        graphene.List,
        child=serializers.IntegerField(min_value=0, max_value=100)
    )

    assert field_a.of_type == graphene.Int

    field_b = assert_conversion(StringListField, graphene.List)

    assert field_b.of_type == graphene.String

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

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

        _type = self.__graphql_type
        if is_repeated:
            _type = List(_type)

        if is_required:
            _type = NonNull(_type)

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

def convert_ndb_scalar_property(graphene_type, ndb_prop, registry=None, **kwargs):
    kwargs['description'] = "%s %s property" % (ndb_prop._name, graphene_type)
    _type = graphene_type

    if ndb_prop._repeated:
        _type = List(_type)

    if ndb_prop._required:
        _type = NonNull(_type)

    return Field(_type, **kwargs)

def testStringProperty_repeated_shouldConvertToList(self):
        ndb_prop = ndb.StringProperty(repeated=True)
        result = convert_ndb_property(ndb_prop)
        graphene_type = result.field._type

        self.assertIsInstance(graphene_type, graphene.List)
        self.assertEqual(graphene_type.of_type, graphene.String)

def testQuery_onlyFields(self):
        Article(headline="h1", summary="s1").put()

        class ArticleType(NdbObjectType):
            class Meta:
                model = Article
                only_fields = ['headline']

        class QueryType(graphene.ObjectType):
            articles = graphene.List(ArticleType)

            def resolve_articles(self, info):
                return Article.query()

        schema = graphene.Schema(query=QueryType)
        query = '''
                    query ArticlesQuery {
                      articles { headline }
                    }
                '''

        result = schema.execute(query)

        self.assertIsNotNone(result.data)
        self.assertEqual(result.data['articles'][0]['headline'], 'h1')

        query = '''
                    query ArticlesQuery {
                      articles { headline, summary }
                    }
                '''
        result = schema.execute(query)

        self.assertIsNotNone(result.errors)
        self.assertTrue('Cannot query field "summary"' in result.errors[0].message)

def test_should_query_well():
    class ReporterType(PynamoObjectType):
        class Meta:
            model = Reporter

    class Query(graphene.ObjectType):
        reporter = graphene.Field(ReporterType)
        reporters = graphene.List(ReporterType)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(1)

        def resolve_reporters(self, *args, **kwargs):
            return list(Reporter.scan())

    query = '''
        query ReporterQuery {
          reporter {
            firstName,
            lastName,
            email,
            customMap,
            awards
          }
          reporters {
            firstName
          }
        }
    '''
    expected = {
        'reporter': {
            'email': None,
            'firstName': 'ABA',
            'lastName': 'X',
            'customMap': {"key1": "value1", "key2": "value2"},
            'awards': ['pulizer']
        },
        'reporters': [{
            'firstName': 'ABO',
        }, {
            'firstName': 'ABA',
        }]
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    result.data['reporter']["customMap"] = json.loads(result.data['reporter']["customMap"])
    assert dict(result.data['reporter']) == expected['reporter']
    assert all(item in result.data['reporters'] for item in expected['reporters'])

def test_should_query_list():
    r1 = Reporter(last_name='ABA')
    r1.save()
    r2 = Reporter(last_name='Griffin')
    r2.save()

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )

    class Query(graphene.ObjectType):
        all_reporters = graphene.List(ReporterType)
        debug = graphene.Field(DjangoDebug, name='__debug')

        def resolve_all_reporters(self, info, **args):
            return Reporter.objects.all()

    query = '''
        query ReporterQuery {
          allReporters {
            lastName
          }
          __debug {
            sql {
              rawSql
            }
          }
        }
    '''
    expected = {
        'allReporters': [{
            'lastName': 'ABA',
        }, {
            'lastName': 'Griffin',
        }],
        '__debug': {
            'sql': [{
                'rawSql': str(Reporter.objects.all().query)
            }]
        }
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()])
    assert not result.errors
    assert result.data == expected

def testNdbObjectType_should_raise_if_model_is_invalid(self):
        with self.assertRaises(Exception) as context:
            class Character2(NdbObjectType):
                class Meta:
                    model = 1

        assert 'not an NDB model' in str(context.exception.message)

    # def testNdbObjectType_keyProperty_kindDoesntExist_raisesException(self):
    #     with self.assertRaises(Exception) as context:
    #         class ArticleType(NdbObjectType):
    #             class Meta:
    #                 model = Article
    #                 only_fields = ('prop',)
    #
    #             prop = NdbKeyReferenceField('foo', 'bar')
    #
    #         class QueryType(graphene.ObjectType):
    #             articles = graphene.List(ArticleType)
    #
    #             @graphene.resolve_only_args
    #             def resolve_articles(self):
    #                 return Article.query()
    #
    #         schema = graphene.Schema(query=QueryType)
    #         schema.execute('query test {  articles { prop } }')
    #
    #     self.assertIn("Model 'bar' is not accessible by the schema.", str(context.exception.message))

    # def testNdbObjectType_keyProperty_stringRepresentation_kindDoesntExist_raisesException(self):
    #     with self.assertRaises(Exception) as context:
    #         class ArticleType(NdbObjectType):
    #             class Meta:
    #                 model = Article
    #                 only_fields = ('prop',)
    #
    #             prop = NdbKeyStringField('foo', 'bar')
    #
    #         class QueryType(graphene.ObjectType):
    #             articles = graphene.List(ArticleType)
    #
    #             @graphene.resolve_only_args
    #             def resolve_articles(self):
    #                 return Article.query()
    #
    #         schema = graphene.Schema(query=QueryType)
    #         schema.execute('query test {  articles { prop } }')
    #
    #     self.assertIn("Model 'bar' is not accessible by the schema.", str(context.exception.message))