Python graphene 模块,Field() 实例源码
我们从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, 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 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_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, contex, info):
return resolver_query(g_schema, 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, *args, **kwargs):
field = django_field(help_text='Custom Help Text', null=True, **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, **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, 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_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, attrs)
options = Options(
attrs.pop('Meta', None),
name=name,
description=attrs.pop('__doc__',
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, options.model)
cls = ObjectTypeMeta.__new__(cls, 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, **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 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, 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
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 __init_subclass_with_Meta__(cls, model=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, (
'The attribute registry in {} needs to be an instance of '
'Registry, 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),
)
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', 'babies'],
'data': '{"angry_babies": true}',
'store': '{"h": "store"}',
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected