Python graphene 模块,Mutation() 实例源码
我们从Python开源项目中,提取了以下9个代码示例,用于说明如何使用graphene.Mutation()。
def test_gen_mutation_user_mutate_wrong_return(schema_builder, mock_person):
import graphene
from graphene_mongo.mutation import gen_mutation
from graphene_mongo.model import ModelSchema
def mutate(args, context):
return False
model_schema = ModelSchema(mock_person, mock_person._fields, mutate, None)
result = gen_mutation(mock_person, model_schema.schema, model_schema.operators_mutation,
model_schema.fields_mutation, model_schema.mutate_func, None)
assert issubclass(result, graphene.Mutation)
assert hasattr(result, 'mutate')
with pytest.raises(Exception) as e_info:
result.mutate(None, {'name': "Test"}, None, None)
assert str(e_info.value) == 'Failed to resolve mutation of the schema {}' \
' because mutate function must return a instance of {},and the return type was {}.'\
.format(model_schema.schema.__name__, model_schema.model.__name__, type(False))
def test_gen_mutation(mock_person):
import inspect
import graphene
from graphene.utils.str_converters import to_snake_case
from graphene_mongo.mutation import gen_mutation
from graphene_mongo.model import ModelSchema
model_schema = ModelSchema(mock_person, 'mutate')
assert result._Meta.name == 'Create' + mock_person.__name__
assert result._Meta.local_fields[to_snake_case(mock_person.__name__)]
assert result._Meta.fields[to_snake_case(mock_person.__name__)]
operators_mutation = inspect.signature(result.__dict__['Field']).parameters['args'].default
assert operators_mutation == model_schema.operators_mutation
def test_gen_mutation_user_mutation_func(mock_person):
import graphene
from graphene_mongo.mutation import gen_mutation
from graphene_mongo.model import ModelSchema
def mutate(args, context):
u = mock_person(**args)
u.save()
return u
model_schema = ModelSchema(mock_person, None)
user_mutate_func = gen_mutation(mock_person, None)
assert issubclass(user_mutate_func, graphene.Mutation)
assert hasattr(user_mutate_func, 'mutate')
assert getattr(user_mutate_func, 'mutate').__name__ == 'user_mutate'
def mutate(cls, root, context, **kwargs):
"""
Mutation logic is handled here
"""
# 'before_insert' hooks won't work with db.session.bulk_save_objects
# we need to find a way to get hooks to work with bulk_save_objects @todo
objs = []
for data in kwargs.get('records', []):
d = Deal.get_object_from_graphql_input(data)
db.session.add(d)
objs.append(d)
try:
db.session.commit()
return cls(ok=True, ids=[obj.id for obj in objs])
except Exception as e:
raise GraphQLError(e.args)
def mutate(cls, **kwargs):
"""
Mutation logic is handled here
"""
# More details about synchronize_session options in sqlAlchemy
# http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.delete
query = Deal.query.filter(
Deal.id.in_(kwargs.get('uids', [])))
objs = []
for obj in query:
db.session.delete(obj)
objs.append(obj)
db.session.info['changes'] = {'created': [], 'updated': [], 'deleted': objs}
try:
db.session.commit()
return cls(ok=True)
except Exception as e:
raise GraphQLError(e.args)
def mutate(cls, []):
c = Contact.get_object_from_graphql_input(data)
db.session.add(c)
objs.append(c)
try:
db.session.commit()
return cls(ok=True, **kwargs):
"""
Mutation logic is handled here
"""
# 'before_insert' hooks won't work with db.session.bulk_save_objects
# we need to find a way to get hooks to work with bulk_save_objects @todo
records = []
for data in kwargs.get('records', []):
actual = Deal.query.get(data['uid'])
if not actual:
raise GraphQLError('Invalid id (%s)' % data['id'])
d = Deal.get_object_from_graphql_input(data)
for column_name, _ in inspect(Deal).attrs.items():
if column_name == 'id':
continue
if column_name not in data:
continue
setattr(actual, column_name, getattr(d, column_name))
db.session.add(actual)
records.append(actual.id)
try:
db.session.commit()
return cls(ok=True, ids=records)
except Exception as e:
raise GraphQLError(e.args)
def mutate(cls, []):
actual = Contact.query.get(data['uid'])
if not actual:
raise GraphQLError('Invalid id (%s)' % data['id'])
c = Contact.get_object_from_graphql_input(data)
for column_name, _ in inspect(Contact).attrs.items():
if column_name == 'id':
continue
if column_name not in data:
continue
setattr(actual, getattr(c, ids=records)
except Exception as e:
raise GraphQLError(e.args)
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, info):
if validator:
validator(model, {}, {})
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, fields_mutation)
return Create(**{to_snake_case(model.__name__): graphene_obj})
def generic_mutate(root, {})
obj = model(**args)
obj.save()
graphene_obj = mongo_to_graphene(obj, 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
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。