我们从Python开源项目中,提取了以下38个代码示例,用于说明如何使用graphene.Field()。
def test_filter_filterset_information_on_meta_related(): class ReporterFilterNode(DjangoObjectType): class Meta: model = Reporter interfaces = (Node, ) filter_fields = ['first_name', 'articles'] class ArticleFilterNode(DjangoObjectType): class Meta: model = Article interfaces = (Node, ) filter_fields = ['headline', 'reporter'] class Query(ObjectType): all_reporters = DjangoFilterConnectionField(ReporterFilterNode) all_articles = DjangoFilterConnectionField(ArticleFilterNode) reporter = Field(ReporterFilterNode) article = Field(ArticleFilterNode) schema = Schema(query=Query) articles_field = ReporterFilterNode._meta.fields['articles'].get_type() assert_arguments(articles_field, 'headline', 'reporter') assert_not_orderable(articles_field)
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 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 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 assert_attribute_conversion(attribute, graphene_field, **kwargs): graphene_type = convert_pynamo_attribute(attribute, attribute) assert isinstance(graphene_type, graphene_field) field = graphene_type.Field() return field
def test_should_onetoone_convert_field(): class A(PynamoObjectType): class Meta: model = Article interfaces = (Node,) dynamic_field = convert_pynamo_attribute(Reporter.favorite_article, Reporter.favorite_article, A._meta.registry) assert isinstance(dynamic_field, Dynamic) graphene_type = dynamic_field.get_type() assert isinstance(graphene_type, graphene.Field) assert graphene_type.type == A
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_root_scan_should_warn_on_params(): class ArticleNode(PynamoObjectType): class Meta: model = Article interfaces = (Node,) class Query(graphene.ObjectType): node = Node.Field() articles = PynamoConnectionField(ArticleNode) query = ''' query ArticlesQuery { articles(after: "QXJ0aWNsZU5vZGU6MQ==") { edges { node { id headline } } } } ''' schema = graphene.Schema(query=Query) result = schema.execute(query) assert result.errors assert isinstance(result.errors[0].original_error, NotImplementedError)
def test_node_replacedfield(): idfield = Human._meta.fields['pub_date'] assert isinstance(idfield, Field) assert idfield.type == Int
def to_attrs(self): return { 'single': self.single(), 'list': self.list(), 'fields': self.fields, 'model': self.model, 'mutate': self.mutate().Field() }
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 reference_field(f_name, mongo_field): """ Generate a schema for RefereceField, or get a schema already done saved in _generated_schemas """ from graphene_mongo import MongoSchema document = mongo_field.document_type_obj # document that this ReferenceField references schema = MongoSchema.get_or_generate_schema(document).schema return graphene.Field(schema)
def assert_orderable(field): args = get_args(field) assert 'order_by' in args, \ 'Field cannot be ordered'
def assert_conversion(django_field, graphene_field, *args, **kwargs): field = django_field(help_text='Custom Help Text', null=True, *args, **kwargs) graphene_type = convert_django_field(field) assert isinstance(graphene_type, graphene_field) field = graphene_type.Field() assert field.description == 'Custom Help Text' nonnull_field = django_field(null=False, *args, **kwargs) if not nonnull_field.null: nonnull_graphene_type = convert_django_field(nonnull_field) nonnull_field = nonnull_graphene_type.Field() assert isinstance(nonnull_field.type, graphene.NonNull) return nonnull_field return field
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_onetoone_reverse_convert_model(): # Django 1.9 uses 'rel', <1.9 uses 'related related = getattr(Film.details, 'rel', None) or \ getattr(Film.details, 'related') class A(DjangoObjectType): class Meta: model = FilmDetails 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 dynamic_field.type == A
def test_should_query_simplelazy_objects(): class ReporterType(DjangoObjectType): class Meta: model = Reporter only_fields = ('id', ) class Query(graphene.ObjectType): reporter = graphene.Field(ReporterType) def resolve_reporter(self, info): return SimpleLazyObject(lambda: Reporter(id=1)) schema = graphene.Schema(query=Query) query = ''' query { reporter { id } } ''' result = schema.execute(query) assert not result.errors assert result.data == { 'reporter': { 'id': '1' } }
def test_should_query_well(): class ReporterType(DjangoObjectType): class Meta: model = Reporter class Query(graphene.ObjectType): reporter = graphene.Field(ReporterType) def resolve_reporter(self, info): return Reporter(first_name='ABA', last_name='X') query = ''' query ReporterQuery { reporter { firstName, lastName, email } } ''' expected = { 'reporter': { 'firstName': 'ABA', 'lastName': 'X', 'email': '' } } schema = graphene.Schema(query=Query) result = schema.execute(query) assert not result.errors assert result.data == expected
def test_should_handle_inherited_choices(): class BaseModel(models.Model): choice_field = models.IntegerField(choices=((0, 'zero'), (1, 'one'))) class ChildModel(BaseModel): class Meta: proxy = True class BaseType(DjangoObjectType): class Meta: model = BaseModel class ChildType(DjangoObjectType): class Meta: model = ChildModel class Query(graphene.ObjectType): base = graphene.Field(BaseType) child = graphene.Field(ChildType) schema = graphene.Schema(query=Query) query = ''' query { child { choiceField } } ''' result = schema.execute(query) assert not result.errors
def test_should_base_field_convert_string(): assert_conversion(serializers.Field, graphene.String)
def test_should_model_convert_field(): class MyModelSerializer(serializers.ModelSerializer): class Meta: model = None fields = '__all__' assert_conversion(MyModelSerializer, graphene.Field, is_input=False)
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 convert_ndb_json_property(ndb_prop, registry=None): return Field(JSONString, description=ndb_prop._name)
def convert_ndb_datetime_property(ndb_prop, registry=None): return Field(DateTime, description=ndb_prop._name)
def make_resolvable_fields(inner_fields): result = {} for name, field in inner_fields.items(): result.update({ name: Field(String, **get_field_args(field)), 'resolve_{}'.format(name): partial(resolve, name), }) return result
def convert_fields_to_graph(fields): for name, field in fields.items(): inner_fields = get_fields(field, callables=True) inner_fields = make_resolvable_fields(inner_fields) field_class = type(name.capitalize(), (ObjectType, ), inner_fields) fields[name] = Field(field_class) return fields
def __new__(cls, name, bases, attrs): # Also ensure initialization is only performed for subclasses of Model # (excluding Model class itself). if not is_base_type(bases, PynamoObjectTypeMeta): return type.__new__(cls, name, bases, attrs) options = Options( attrs.pop('Meta', None), name=name, description=attrs.pop('__doc__', None), model=None, local_fields=None, only_fields=(), exclude_fields=(), id='id', interfaces=(), registry=None ) if not options.registry: options.registry = get_global_registry() assert isinstance(options.registry, Registry), ( 'The attribute registry in {}.Meta needs to be an' ' instance of Registry, received "{}".' ).format(name, options.registry) assert (inspect.isclass(options.model) and issubclass(options.model, Model)), ( 'You need to pass a valid PynamoDB Model in ' '{}.Meta, received "{}".' ).format(name, options.model) cls = ObjectTypeMeta.__new__(cls, name, bases, dict(attrs, _meta=options)) options.registry.register(cls) options.pynamo_fields = yank_fields_from_attrs( construct_fields(options), _as=Field, ) options.fields = merge( options.interface_fields, options.pynamo_fields, options.base_fields, options.local_fields ) return cls
def test_should_custom_identifier(): class EditorNode(PynamoObjectType): class Meta: model = Editor interfaces = (Node,) class Query(graphene.ObjectType): node = Node.Field() all_editors = PynamoConnectionField(EditorNode) query = ''' query EditorQuery { allEditors { edges { node { id, name } } }, node(id: "RWRpdG9yTm9kZTox") { ...on EditorNode { name } } } ''' expected = { 'allEditors': { 'edges': [{ 'node': { 'id': 'RWRpdG9yTm9kZTox', 'name': 'John' } }] }, 'node': { 'name': 'John' } } schema = graphene.Schema(query=Query) result = schema.execute(query) assert not result.errors assert result.data['allEditors'] == expected['allEditors']
def test_should_return_empty_cursors_on_empty(): class ArticleNode(PynamoObjectType): class Meta: model = Article interfaces = (Node,) class ReporterNode(PynamoObjectType): class Meta: model = Reporter interfaces = (Node,) class Query(graphene.ObjectType): node = Node.Field() reporter = graphene.Field(ReporterNode) def resolve_reporter(self, *args, **kwargs): return Reporter.get(2) query = ''' query ReporterQuery { reporter { id, firstName, articles(first: 1) { edges { node { id headline } } pageInfo { hasNextPage hasPreviousPage startCursor endCursor } } lastName, email } } ''' expected = { 'reporter': { 'articles': { 'edges': [], 'pageInfo': { 'hasNextPage': False, 'hasPreviousPage': False, 'startCursor': '', 'endCursor': '' } } } } schema = graphene.Schema(query=Query) result = schema.execute(query) assert not result.errors assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges'] assert result.data['reporter']['articles']['pageInfo'] == expected['reporter']['articles']['pageInfo']
def test_should_support_last(): class ArticleNode(PynamoObjectType): class Meta: model = Article interfaces = (Node,) class ReporterNode(PynamoObjectType): class Meta: model = Reporter interfaces = (Node,) class Query(graphene.ObjectType): node = Node.Field() reporter = graphene.Field(ReporterNode) def resolve_reporter(self, *args, **kwargs): return Reporter.get(1) query = ''' query ReporterQuery { reporter { id, firstName, articles(last: 1) { edges { node { id headline } } } lastName, email } } ''' expected = { 'reporter': { 'articles': { 'edges': [{ 'node': { 'id': 'QXJ0aWNsZU5vZGU6Mw==', 'headline': 'My Article' } }] } } } schema = graphene.Schema(query=Query) result = schema.execute(query) assert not result.errors assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']
def test_should_support_after(): class ArticleNode(PynamoObjectType): class Meta: model = Article interfaces = (Node,) class ReporterNode(PynamoObjectType): class Meta: model = Reporter interfaces = (Node,) class Query(graphene.ObjectType): node = Node.Field() reporter = graphene.Field(ReporterNode) def resolve_reporter(self, *args, **kwargs): return Reporter.get(1) query = ''' query ReporterQuery { reporter { id, firstName, articles(after: "QXJ0aWNsZU5vZGU6MQ==") { edges { node { id headline } } } lastName, email } } ''' expected = { 'reporter': { 'articles': { 'edges': [{ 'node': { 'id': 'QXJ0aWNsZU5vZGU6Mw==', 'headline': 'My Article' } }] } } } schema = graphene.Schema(query=Query) result = schema.execute(query) assert not result.errors assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']
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
def test_should_query_field(): 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): reporter = graphene.Field(ReporterType) debug = graphene.Field(DjangoDebug, name='__debug') def resolve_reporter(self, info, **args): return Reporter.objects.first() query = ''' query ReporterQuery { reporter { lastName } __debug { sql { rawSql } } } ''' expected = { 'reporter': { 'lastName': 'ABA', }, '__debug': { 'sql': [{ 'rawSql': str(Reporter.objects.order_by('pk')[:1].query) }] } } schema = graphene.Schema(query=Query) result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()]) assert not result.errors assert result.data == expected
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 test_should_query_connection(): 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 = DjangoConnectionField(ReporterType) debug = graphene.Field(DjangoDebug, name='__debug') def resolve_all_reporters(self, info, **args): return Reporter.objects.all() query = ''' query ReporterQuery { allReporters(first:1) { edges { node { lastName } } } __debug { sql { rawSql } } } ''' expected = { 'allReporters': { 'edges': [{ 'node': { 'lastName': 'ABA', } }] }, } schema = graphene.Schema(query=Query) result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()]) assert not result.errors assert result.data['allReporters'] == expected['allReporters'] assert 'COUNT' in result.data['__debug']['sql'][0]['rawSql'] query = str(Reporter.objects.all()[:1].query) assert result.data['__debug']['sql'][1]['rawSql'] == query
def test_should_query_connectionfilter(): from ...filter import DjangoFilterConnectionField 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 = DjangoFilterConnectionField(ReporterType, fields=['last_name']) s = graphene.String(resolver=lambda *_: "S") debug = graphene.Field(DjangoDebug, name='__debug') def resolve_all_reporters(self, info, **args): return Reporter.objects.all() query = ''' query ReporterQuery { allReporters(first:1) { edges { node { lastName } } } __debug { sql { rawSql } } } ''' expected = { 'allReporters': { 'edges': [{ 'node': { 'lastName': 'ABA', } }] }, } schema = graphene.Schema(query=Query) result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()]) assert not result.errors assert result.data['allReporters'] == expected['allReporters'] assert 'COUNT' in result.data['__debug']['sql'][0]['rawSql'] query = str(Reporter.objects.all()[:1].query) assert result.data['__debug']['sql'][1]['rawSql'] == query
def __init_subclass_with_meta__(cls, model=None, registry=None, skip_registry=False, only_fields=(), exclude_fields=(), filter_fields=None, connection=None, connection_class=None, use_connection=None, interfaces=(), **options): assert is_valid_django_model(model), ( 'You need to pass a valid Django Model in {}.Meta, received "{}".' ).format(cls.__name__, model) if not registry: registry = get_global_registry() assert isinstance(registry, Registry), ( 'The attribute registry in {} needs to be an instance of ' 'Registry, received "{}".' ).format(cls.__name__, registry) if not DJANGO_FILTER_INSTALLED and filter_fields: raise Exception("Can only set filter_fields if Django-Filter is installed") django_fields = yank_fields_from_attrs( construct_fields(model, registry, only_fields, exclude_fields), _as=Field, ) if use_connection is None and interfaces: use_connection = any((issubclass(interface, Node) for interface in interfaces)) if use_connection and not connection: # We create the connection automatically if not connection_class: connection_class = Connection connection = connection_class.create_type( '{}Connection'.format(cls.__name__), node=cls) if connection is not None: assert issubclass(connection, Connection), ( "The connection must be a Connection. Received {}" ).format(connection.__name__) _meta = DjangoObjectTypeOptions(cls) _meta.model = model _meta.registry = registry _meta.filter_fields = filter_fields _meta.fields = django_fields _meta.connection = connection super(DjangoObjectType, cls).__init_subclass_with_meta__(_meta=_meta, interfaces=interfaces, **options) if not skip_registry: registry.register(cls)
def test_should_query_postgres_fields(): from django.contrib.postgres.fields import IntegerRangeField, ArrayField, JSONField, HStoreField class Event(models.Model): ages = IntegerRangeField(help_text='The age ranges') data = JSONField(help_text='Data') store = HStoreField() tags = ArrayField(models.CharField(max_length=50)) class EventType(DjangoObjectType): class Meta: model = Event class Query(graphene.ObjectType): event = graphene.Field(EventType) def resolve_event(self, info): return Event( ages=(0, 10), data={'angry_babies': True}, store={'h': 'store'}, tags=['child', 'angry', 'babies'] ) schema = graphene.Schema(query=Query) query = ''' query myQuery { event { ages tags data store } } ''' expected = { 'event': { 'ages': [0, 10], 'tags': ['child', 'angry', 'babies'], 'data': '{"angry_babies": true}', 'store': '{"h": "store"}', }, } result = schema.execute(query) assert not result.errors assert result.data == expected