Python graphene 模块,Schema() 实例源码
我们从Python开源项目中,提取了以下33个代码示例,用于说明如何使用graphene.Schema()。
def query_route(request):
"""
Query database based on Facebook's GraphQL Library.
Parameters must be coded into a "q" GET parameter,e.g.
`curl "localhost:4284/api/v2/query?q=query\{statements\{uid,isstartpoint\}\}"`
:return: JSON containing queried data
"""
q = request.params.get("q")
if q:
schema = graphene.Schema(query=Query)
result = schema.execute(q, context_value={'session': DBdiscussionSession})
if result.errors:
return {"errors": {"message": "Not all requested parameters Could be queried. Some fields are not "
"allowed,e.g. the password.",
"exception": str(result.errors)}}
return result.data
return {"errors": {"message": "No valid query provided."}}
def test_should_query_only_fields():
with raises(Exception):
class ReporterType(DjangoObjectType):
class Meta:
model = Reporter
only_fields = ('articles', )
schema = graphene.Schema(query=ReporterType)
query = '''
query ReporterQuery {
articles
}
'''
result = schema.execute(query)
assert not result.errors
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 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 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', attrs)
return graphene.Schema(query=Query, mutation=Mutation)
return build
def graphql_schema(self):
"""
:return: Graphql Schema
:rtype: graphene.Schema
"""
return self._graphql_schema
def init_graphql_schema():
"""
Go through all sub apps defined Queries,Types and mutations
defined in (graphql.py) and register them in one global schema
"""
# Import all (graphql) submodules defined in package 'graphql'
# After importing we'll have
# All Queries under :: BaseQuery.__subclasses__()
# All Types under :: sqlAlchemyObjectType.__subclasses__()
# All Mutations under :: BaseMutation.__subclasses__()
CRM._load_modules(module_type='types')
CRM._load_modules(module_type='queries')
CRM._load_modules(module_type='mutations')
schema = graphene.Schema(
# Make dynamic Query class that inherits all defined queries
query=type(
'Query',
tuple(BaseQuery.__subclasses__()),
{}
),
types=list(sqlAlchemyObjectType.__subclasses__()),
# Make dynamic Mutations class that inherits all defined mutations
mutation=type(
'Mutations',
tuple(BaseMutation.__subclasses__()),
{}
)
)
return schema
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 __init__(self, app=None, blueprint_options=None):
self._schema = graphene.Schema()
self._webpack = None
if blueprint_options is None:
blueprint_options = {}
self._blueprint_options = blueprint_options
self._blueprint = self._create_blueprint()
self._add_url_rules()
self.jwt = None
if app is not None:
self.init_app(app)
def test_noSchema_returns500(self):
graphql_application = webapp2.WsgiApplication([
('/graphql', GraphQLHandler)
])
app = webtest.TestApp(graphql_application)
for method in (app.get, app.post):
response = method('/graphql', expect_errors=True)
self.assertEqual(response.status_int, 500)
self.assertEqual(response.json_body['errors'][0]['message'], 'GraphQL Schema is missing.')
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 router(app):
schema = graphene.Schema(query=RootQuery)
router = GrapheneRouter.from_schema(schema).serving_on(app.app)
return router
def __init__(self, schema: graphene.Schema):
self.schema = schema
def from_schema(cls, schema: graphene.Schema) -> "GrapheneRouter":
"""Create a router with a Graphene schema.
Args:
schema: A :class:`graphene.Schema` instance.
Returns:
A :class:`.GrapheneRouter` set up with the given schema.
"""
router = GrapheneRouter()
return router.with_schema(schema)
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 举报,一经查实,本站将立刻删除。