我们从Python开源项目中,提取了以下30个代码示例,用于说明如何使用graphene.ObjectType()。
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 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 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 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 test_root_scan_should_fastforward_on_after(): class ArticleNode(PynamoObjectType): class Meta: model = Article interfaces = (Node,) class Query(graphene.ObjectType): node = Node.Field() articles = PynamoConnectionField(ArticleNode) def resolve_articles(self, *args, **kwargs): return [ Article(1, headline='One'), Article(2, headline='Two'), Article(3, headline='Three'), Article(4, headline='Four') ] query = ''' query ArticlesQuery { articles(after: "QXJ0aWNsZU5vZGU6Mq==", first: 1) { edges { node { id headline } } } } ''' expected = [{ 'node': { 'headline': 'Three', 'id': 'QXJ0aWNsZU5vZGU6Mw==' } }] schema = graphene.Schema(query=Query) result = schema.execute(query) assert not result.errors assert result.data['articles']['edges'] == expected
def test_objecttype_registered(): assert issubclass(Character, ObjectType) assert Character._meta.model == Reporter assert list(Character._meta.fields.keys()) == [ 'articles', 'awards', 'custom_map', 'email', 'favorite_article', 'first_name', 'id', 'last_name', 'pets']
def test_object_type(): assert issubclass(Human, ObjectType) assert list(Human._meta.fields.keys()) == ['id', 'headline', 'reporter', 'pub_date'] assert is_node(Human)
def __init__(self, model, attrs_mongo_doc, mutate, validator): self.model = model self.mutate_func = mutate # mutate function that user specified self.validator = validator self.fields, self.fields_mutation, self.operators_mutation,\ self.operators_single, self.operators_list = convert_fields(attrs_mongo_doc) self.schema = type(self.model.__name__ + 'Graphene', (graphene.ObjectType,), self.fields.copy())
def schema_builder(): def build(schemas, mutations=None): mutations = [] if not mutations else mutations attrs = {schema[0].model.__name__.lower(): schema[1] for schema in schemas} Query = type('Query', (graphene.ObjectType,), attrs) attrs = {'create_' + m.model.__name__.lower(): m.mutate for m in mutations} Mutation = type('Mutation', (graphene.ObjectType,), attrs) return graphene.Schema(query=Query, mutation=Mutation) return build
def test_recursive_filter_connection(): class ReporterFilterNode(DjangoObjectType): child_reporters = DjangoFilterConnectionField(lambda: ReporterFilterNode) def resolve_child_reporters(self, **args): return [] class Meta: model = Reporter interfaces = (Node, ) class Query(ObjectType): all_reporters = DjangoFilterConnectionField(ReporterFilterNode) assert ReporterFilterNode._meta.fields['child_reporters'].node_type == ReporterFilterNode
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_query_promise_connectionfields(): from promise import Promise class ReporterType(DjangoObjectType): class Meta: model = Reporter interfaces = (Node, ) class Query(graphene.ObjectType): all_reporters = DjangoConnectionField(ReporterType) def resolve_all_reporters(self, info, **args): return Promise.resolve([Reporter(id=1)]) schema = graphene.Schema(query=Query) query = ''' query ReporterPromiseConnectionQuery { allReporters(first: 1) { edges { node { id } } } } ''' expected = { 'allReporters': { 'edges': [{ 'node': { 'id': 'UmVwb3J0ZXJUeXBlOjE=' } }] } } 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 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 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 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_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 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 test_annotation_is_perserved(): class ReporterType(DjangoObjectType): full_name = String() def resolve_full_name(instance, info, **args): return instance.full_name class Meta: model = Reporter interfaces = (Node, ) filter_fields = () class Query(ObjectType): all_reporters = DjangoFilterConnectionField(ReporterType) def resolve_all_reporters(self, info, **args): return Reporter.objects.annotate( full_name=Concat('first_name', Value(' '), 'last_name', output_field=TextField()) ) Reporter.objects.create( first_name='John', last_name='Doe', ) schema = Schema(query=Query) query = ''' query NodeFilteringQuery { allReporters(first: 1) { edges { node { fullName } } } } ''' expected = { 'allReporters': { 'edges': [{ 'node': { 'fullName': 'John Doe', } }] } } result = schema.execute(query) assert not result.errors assert result.data == expected
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
def test_should_query_connectionfields(): class ReporterType(DjangoObjectType): class Meta: model = Reporter interfaces = (Node, ) only_fields = ('articles', ) class Query(graphene.ObjectType): all_reporters = DjangoConnectionField(ReporterType) def resolve_all_reporters(self, info, **args): return [Reporter(id=1)] schema = graphene.Schema(query=Query) query = ''' query ReporterConnectionQuery { allReporters { pageInfo { hasNextPage } edges { node { id } } } } ''' result = schema.execute(query) assert not result.errors assert result.data == { 'allReporters': { 'pageInfo': { 'hasNextPage': False, }, 'edges': [{ 'node': { 'id': 'UmVwb3J0ZXJUeXBlOjE=' } }] } }
def test_should_keep_annotations(): from django.db.models import ( Count, Avg, ) class ReporterType(DjangoObjectType): class Meta: model = Reporter interfaces = (Node, ) only_fields = ('articles', ) class ArticleType(DjangoObjectType): class Meta: model = Article interfaces = (Node, ) filter_fields = ('lang', ) class Query(graphene.ObjectType): all_reporters = DjangoConnectionField(ReporterType) all_articles = DjangoConnectionField(ArticleType) def resolve_all_reporters(self, info, **args): return Reporter.objects.annotate(articles_c=Count('articles')).order_by('articles_c') def resolve_all_articles(self, info, **args): return Article.objects.annotate(import_avg=Avg('importance')).order_by('import_avg') schema = graphene.Schema(query=Query) query = ''' query ReporterConnectionQuery { allReporters { pageInfo { hasNextPage } edges { node { id } } } allArticles { pageInfo { hasNextPage } edges { node { id } } } } ''' result = schema.execute(query) assert not result.errors
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))