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
python类List()的实例源码
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)