Python graphene 模块,ObjectType() 实例源码
我们从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, info):
return self
def resolve_test_context(self, info):
return context
def resolve_test_filter(self, info):
return 'good_filter' if args.get('filterBoolean') else 'bad_filter'
def resolve_test_filter_multi(self, info):
return 'good_filter' if args.get('filterBoolean') else 'bad_filter'
def resolve_test_channel_options(self, 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, info):
id = args['id']
name = data[args['id']]['name']
return UserType(id=id, name=name)
def resolve_user_filtered(self, name=name)
def resolve_context(self, info):
return context
def resolve_error(self, info):
raise Exception('E1')
return graphene.Schema(query=Query, subscription=Subscription)
def resolve_address(self, 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_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', attrs)
attrs = {'create_' + m.model.__name__.lower(): m.mutate for m in mutations}
Mutation = type('Mutation', 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, **kwargs):
return Reporter.get(1)
def resolve_reporters(self, **kwargs):
return list(Reporter.scan())
query = '''
query ReporterQuery {
reporter {
firstName,
email,
customMap,
awards
}
reporters {
firstName
}
}
'''
expected = {
'reporter': {
'email': None,
'firstName': 'ABA',
'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, **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, **kwargs):
return Reporter.get(1)
query = '''
query ReporterQuery {
reporter {
id,
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,
articles(after: "QXJ0aWNsZU5vZGU6MQ==") {
edges {
node {
id
headline
}
}
}
lastName,
'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, **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, **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, 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, **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, 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, 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, **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, **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', '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', **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, )
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, **args):
return Reporter.objects.annotate(articles_c=Count('articles')).order_by('articles_c')
def resolve_all_articles(self, **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',str(context.exception.message))
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。