Merge remote-tracking branch 'graphql-python/master'

This commit is contained in:
Jake 2015-10-10 20:07:45 -04:00
commit a59afd4bfa
60 changed files with 3003 additions and 211 deletions

View File

@ -2,10 +2,13 @@ language: python
sudo: false
python:
- 2.7
- 3.3
- 3.4
- 3.5
- pypy
install:
- pip install pytest pytest-cov coveralls flake8 six
- pip install git+https://github.com/dittos/graphqllib.git # Last version of graphqllib
- pip install graphql-relay
- pip install pytest pytest-cov coveralls flake8 six blinker pytest-django
- pip install -e .[django]
- python setup.py develop
script:
- py.test --cov=graphene

View File

@ -1,10 +1,20 @@
# Graphene: GraphQL Object Mapper
# ![Graphene Logo](http://graphene-python.org/favicon.png) [Graphene](http://graphene-python.org) [![Build Status](https://travis-ci.org/graphql-python/graphene.svg?branch=master)](https://travis-ci.org/graphql-python/graphene) [![Coverage Status](https://coveralls.io/repos/graphql-python/graphene/badge.svg?branch=master&service=github)](https://coveralls.io/github/graphql-python/graphene?branch=master)
This is a library to use GraphQL in Python in a easy way.
It will map the models/fields to internal GraphQL-py objects without effort.
[![Build Status](https://travis-ci.org/syrusakbary/graphene.svg?branch=master)](https://travis-ci.org/syrusakbary/graphene)
[![Coverage Status](https://coveralls.io/repos/syrusakbary/graphene/badge.svg?branch=master&service=github)](https://coveralls.io/github/syrusakbary/graphene?branch=master)
Graphene is a Python library for building GraphQL schemas/types fast and easily.
* **Easy to use:** It maps the models/fields to internal GraphQL objects without effort.
* **Relay:** Graphene has builtin support for Relay
* **Django:** Automatic [Django models](#djangorelay-schema) conversion. *See an [example Django](http://github.com/graphql-python/swapi-graphene) implementation*
## Installation
For instaling graphene, just run this command in your shell
```bash
pip install graphene
```
## Usage
@ -13,58 +23,24 @@ Example code of a GraphQL schema using Graphene:
### Schema definition
```python
import graphene
# ...
class Character(graphene.Interface):
id = graphene.IDField()
name = graphene.StringField()
friends = graphene.ListField('self')
appearsIn = graphene.ListField(Episode)
def resolve_friends(self, args, *_):
return [wrap_character(getCharacter(f)) for f in self.instance.friends]
return [Human(f) for f in self.instance.friends]
class Human(Character):
homePlanet = graphene.StringField()
class Droid(Character):
primaryFunction = graphene.StringField()
class Query(graphene.ObjectType):
hero = graphene.Field(Character,
episode = graphene.Argument(Episode)
)
human = graphene.Field(Human,
id = graphene.Argument(graphene.String)
)
droid = graphene.Field(Droid,
id = graphene.Argument(graphene.String)
)
human = graphene.Field(Human)
@resolve_only_args
def resolve_hero(self, episode):
return wrap_character(getHero(episode))
@resolve_only_args
def resolve_human(self, id):
return wrap_character(getHuman(id))
if human:
return Human(human)
@resolve_only_args
def resolve_droid(self, id):
return wrap_character(getDroid(id))
Schema = graphene.Schema(query=Query)
schema = graphene.Schema(query=Query)
```
### Querying
Querying `graphene.Schema` is as simple as:
Then Querying `graphene.Schema` is as simple as:
```python
query = '''
@ -74,7 +50,41 @@ query = '''
}
}
'''
result = Schema.execute(query)
result = schema.execute(query)
```
### Relay Schema
Graphene also supports Relay, check the [Starwars Relay example](tests/starwars_relay)!
```python
class Ship(relay.Node):
name = graphene.StringField()
@classmethod
def get_node(cls, id):
return Ship(your_ship_instance)
class Query(graphene.ObjectType):
ships = relay.ConnectionField(Ship)
node = relay.NodeField()
```
### Django+Relay Schema
If you want to use graphene with your Django Models check the [Starwars Django example](tests/starwars_django)!
```python
class Ship(DjangoNode):
class Meta:
model = YourDjangoModelHere
# only_fields = ('id', 'name') # Only map this fields from the model
# exclude_fields ('field_to_exclude', ) # Exclude mapping this fields from the model
class Query(graphene.ObjectType):
node = relay.NodeField()
```
## Contributing

122
README.rst Normal file
View File

@ -0,0 +1,122 @@
Graphene |Build Status| |Coverage Status|
=========================================
Graphene is a Python library for creating GraphQL schemas/types easly.
It maps the models/fields to internal GraphQL objects without effort.
Including automatic `Django models`_ conversion.
Installation
------------
For instaling graphene, just run this command in your shell
.. code:: bash
pip install graphene
Usage
-----
Example code of a GraphQL schema using Graphene:
Schema definition
~~~~~~~~~~~~~~~~~
.. code:: python
class Character(graphene.Interface):
id = graphene.IDField()
name = graphene.StringField()
friends = graphene.ListField('self')
def resolve_friends(self, args, *_):
return [Human(f) for f in self.instance.friends]
class Human(Character):
homePlanet = graphene.StringField()
class Query(graphene.ObjectType):
human = graphene.Field(Human)
schema = graphene.Schema(query=Query)
Querying
~~~~~~~~
Querying ``graphene.Schema`` is as simple as:
.. code:: python
query = '''
query HeroNameQuery {
hero {
name
}
}
'''
result = schema.execute(query)
Relay Schema
~~~~~~~~~~~~
Graphene also supports Relay, check the `Starwars Relay example`_!
.. code:: python
class Ship(relay.Node):
'''A ship in the Star Wars saga'''
name = graphene.StringField(description='The name of the ship.')
@classmethod
def get_node(cls, id):
return Ship(getShip(id))
class Query(graphene.ObjectType):
ships = relay.ConnectionField(Ship, description='The ships used by the faction.')
node = relay.NodeField()
@resolve_only_args
def resolve_ships(self):
return [Ship(s) for s in getShips()]
Django+Relay Schema
~~~~~~~~~~~~~~~~~~~
If you want to use graphene with your Django Models check the `Starwars
Django example`_!
.. code:: python
class Ship(DjangoNode):
class Meta:
model = YourDjangoModelHere
# only_fields = ('id', 'name') # Only map this fields from the model
# excluxe_fields ('field_to_excluxe', ) # Exclude mapping this fields from the model
class Query(graphene.ObjectType):
node = relay.NodeField()
Contributing
------------
After cloning this repo, ensure dependencies are installed by running:
.. code:: sh
python setup.py install
After developing, the full test suite can be evaluated by running:
.. code:: sh
python setup.py test # Use --pytest-args="-v -s" for verbose mode
.. _Django models: #djangorelay-schema
.. _Starwars Relay example: tests/starwars_relay
.. _Starwars Django example: tests/starwars_django
.. |Build Status| image:: https://travis-ci.org/graphql-python/graphene.svg?branch=master
:target: https://travis-ci.org/graphql-python/graphene
.. |Coverage Status| image:: https://coveralls.io/repos/graphql-python/graphene/badge.svg?branch=master&service=github
:target: https://coveralls.io/github/graphql-python/graphene?branch=master

View File

@ -1,12 +1,22 @@
from graphql.core.type import (
GraphQLEnumType as Enum,
GraphQLArgument as Argument,
# GraphQLSchema as Schema,
GraphQLString as String,
GraphQLInt as Int,
GraphQLID as ID
)
from graphene import signals
from graphene.core.schema import (
Schema
)
from graphene.core.types import (
ObjectType,
Interface
)
from graphene.core.fields import (
Field,
StringField,
@ -15,14 +25,11 @@ from graphene.core.fields import (
IDField,
ListField,
NonNullField,
)
from graphene.core.types import (
ObjectType,
Interface,
Schema
FloatField,
)
from graphene.decorators import (
resolve_only_args
)
# import graphene.relay

View File

View File

@ -0,0 +1,8 @@
from graphene.contrib.django.types import (
DjangoObjectType,
DjangoNode
)
from graphene.contrib.django.fields import (
DjangoConnectionField,
DjangoModelField
)

View File

@ -0,0 +1,71 @@
from singledispatch import singledispatch
from django.db import models
from graphene.core.fields import (
StringField,
IDField,
IntField,
BooleanField,
FloatField,
ListField
)
from graphene.contrib.django.fields import ConnectionOrListField, DjangoModelField
@singledispatch
def convert_django_field(field):
raise Exception(
"Don't know how to convert the Django field %s (%s)" % (field, field.__class__))
@convert_django_field.register(models.DateField)
@convert_django_field.register(models.CharField)
@convert_django_field.register(models.TextField)
@convert_django_field.register(models.EmailField)
@convert_django_field.register(models.SlugField)
@convert_django_field.register(models.URLField)
@convert_django_field.register(models.UUIDField)
def _(field):
return StringField(description=field.help_text)
@convert_django_field.register(models.AutoField)
def _(field):
return IDField(description=field.help_text)
@convert_django_field.register(models.PositiveIntegerField)
@convert_django_field.register(models.PositiveSmallIntegerField)
@convert_django_field.register(models.SmallIntegerField)
@convert_django_field.register(models.BigIntegerField)
@convert_django_field.register(models.IntegerField)
def _(field):
return IntField(description=field.help_text)
@convert_django_field.register(models.BooleanField)
def _(field):
return BooleanField(description=field.help_text, required=True)
@convert_django_field.register(models.NullBooleanField)
def _(field):
return BooleanField(description=field.help_text)
@convert_django_field.register(models.FloatField)
def _(field):
return FloatField(description=field.help_text)
@convert_django_field.register(models.ManyToManyField)
@convert_django_field.register(models.ManyToOneRel)
def _(field):
model_field = DjangoModelField(field.related_model)
return ConnectionOrListField(model_field)
@convert_django_field.register(models.OneToOneField)
@convert_django_field.register(models.ForeignKey)
def _(field):
return DjangoModelField(field.related_model, description=field.help_text)

View File

@ -0,0 +1,92 @@
from graphene.core.fields import (
ListField
)
from graphene import relay
from graphene.core.fields import Field, LazyField
from graphene.utils import cached_property, memoize, LazyMap
from graphene.relay.types import BaseNode
from django.db.models.query import QuerySet
from django.db.models.manager import Manager
@memoize
def get_type_for_model(schema, model):
schema = schema
types = schema.types.values()
for _type in types:
type_model = hasattr(_type, '_meta') and getattr(
_type._meta, 'model', None)
if model == type_model:
return _type
def lazy_map(value, func):
if isinstance(value, Manager):
value = value.get_queryset()
if isinstance(value, QuerySet):
return LazyMap(value, func)
return value
class DjangoConnectionField(relay.ConnectionField):
def wrap_resolved(self, value, instance, args, info):
schema = info.schema.graphene_schema
return lazy_map(value, self.get_object_type(schema))
class LazyListField(ListField):
def resolve(self, instance, args, info):
schema = info.schema.graphene_schema
resolved = super(LazyListField, self).resolve(instance, args, info)
return lazy_map(resolved, self.get_object_type(schema))
class ConnectionOrListField(LazyField):
@memoize
def get_field(self, schema):
model_field = self.field_type
field_object_type = model_field.get_object_type(schema)
if field_object_type and issubclass(field_object_type, BaseNode):
field = DjangoConnectionField(model_field)
else:
field = LazyListField(model_field)
field.contribute_to_class(self.object_type, self.name)
return field
class DjangoModelField(Field):
def __init__(self, model, *args, **kwargs):
super(DjangoModelField, self).__init__(None, *args, **kwargs)
self.model = model
def resolve(self, instance, args, info):
resolved = super(DjangoModelField, self).resolve(instance, args, info)
schema = info.schema.graphene_schema
_type = self.get_object_type(schema)
assert _type, ("Field %s cannot be retrieved as the "
"ObjectType is not registered by the schema" % (
self.field_name
))
return _type(resolved)
@memoize
def internal_type(self, schema):
_type = self.get_object_type(schema)
if not _type and self.object_type._meta.only_fields:
raise Exception(
"Model %r is not accessible by the schema. "
"You can either register the type manually "
"using @schema.register. "
"Or disable the field %s in %s" % (
self.model,
self.field_name,
self.object_type
)
)
return _type and _type.internal_type(schema) or Field.SKIP
def get_object_type(self, schema):
return get_type_for_model(schema, self.model)

View File

@ -0,0 +1,33 @@
import inspect
from django.db import models
from graphene.core.options import Options
from graphene.core.types import BaseObjectType
from graphene.relay.utils import is_node
VALID_ATTRS = ('model', 'only_fields', 'exclude_fields')
def is_base(cls):
from graphene.contrib.django.types import DjangoObjectType
return DjangoObjectType in cls.__bases__
class DjangoOptions(Options):
def __init__(self, *args, **kwargs):
self.model = None
super(DjangoOptions, self).__init__(*args, **kwargs)
self.valid_attrs += VALID_ATTRS
self.only_fields = None
self.exclude_fields = []
def contribute_to_class(self, cls, name):
super(DjangoOptions, self).contribute_to_class(cls, name)
if not is_node(cls) and not is_base(cls):
return
if not self.model:
raise Exception(
'Django ObjectType %s must have a model in the Meta class attr' % cls)
elif not inspect.isclass(self.model) or not issubclass(self.model, models.Model):
raise Exception('Provided model in %s is not a Django model' % cls)

View File

@ -0,0 +1,56 @@
import six
from django.db import models
from graphene.core.types import ObjectTypeMeta, BaseObjectType
from graphene.contrib.django.options import DjangoOptions
from graphene.contrib.django.converter import convert_django_field
from graphene.relay.types import Node, BaseNode
def get_reverse_fields(model):
for name, attr in model.__dict__.items():
related = getattr(attr, 'related', None)
if isinstance(related, models.ManyToOneRel):
yield related
class DjangoObjectTypeMeta(ObjectTypeMeta):
options_cls = DjangoOptions
def is_interface(cls, parents):
return DjangoInterface in parents
def add_extra_fields(cls):
if not cls._meta.model:
return
only_fields = cls._meta.only_fields
reverse_fields = get_reverse_fields(cls._meta.model)
all_fields = sorted(list(cls._meta.model._meta.fields) +
list(cls._meta.model._meta.local_many_to_many))
all_fields += list(reverse_fields)
for field in all_fields:
is_not_in_only = only_fields and field.name not in only_fields
is_excluded = field.name in cls._meta.exclude_fields
if is_not_in_only or is_excluded:
# We skip this field if we specify only_fields and is not
# in there. Or when we excldue this field in exclude_fields
continue
converted_field = convert_django_field(field)
cls.add_to_class(field.name, converted_field)
class DjangoObjectType(six.with_metaclass(DjangoObjectTypeMeta, BaseObjectType)):
pass
class DjangoInterface(six.with_metaclass(DjangoObjectTypeMeta, BaseObjectType)):
pass
class DjangoNode(BaseNode, DjangoInterface):
@classmethod
def get_node(cls, id):
instance = cls._meta.model.objects.filter(id=id).first()
return cls(instance)

View File

@ -0,0 +1,66 @@
import json
from django.http import JsonResponse
from django.views.generic import View
from django.conf import settings
from graphql.core.error import GraphQLError, format_error
def form_error(error):
if isinstance(error, GraphQLError):
return format_error(error)
return error
class GraphQLView(View):
schema = None
@staticmethod
def format_result(result):
data = {'data': result.data}
if result.errors:
data['errors'] = list(map(form_error, result.errors))
return data
def response_errors(self, *errors):
return JsonResponse({
"errors": [{
"message": str(e)
} for e in errors]
})
def execute_query(self, request, query):
if not query:
return self.response_errors(Exception("Must provide query string."))
else:
try:
result = self.schema.execute(query, root=object())
data = self.format_result(result)
except Exception as e:
if settings.DEBUG:
raise e
return self.response_errors(e)
return JsonResponse(data)
def get(self, request, *args, **kwargs):
query = request.GET.get('query')
return self.execute_query(request, query or '')
@staticmethod
def get_content_type(request):
meta = request.META
return meta.get('CONTENT_TYPE', meta.get('HTTP_CONTENT_TYPE', ''))
def post(self, request, *args, **kwargs):
content_type = self.get_content_type(request)
if content_type == 'application/json':
try:
received_json_data = json.loads(request.body.decode())
query = received_json_data.get('query')
except ValueError:
return self.response_errors(ValueError("Malformed json body in the post data"))
else:
query = request.POST.get('query') or request.GET.get('query')
return self.execute_query(request, query or '')

View File

@ -1,4 +1,6 @@
import inspect
import six
from functools import total_ordering
from graphql.core.type import (
GraphQLField,
GraphQLList,
@ -8,70 +10,87 @@ from graphql.core.type import (
GraphQLBoolean,
GraphQLID,
GraphQLArgument,
GraphQLFloat,
)
from graphene.core.types import ObjectType, Interface
from graphene.utils import cached_property
from graphene.utils import memoize, to_camel_case
from graphene.core.types import BaseObjectType
from graphene.core.scalars import GraphQLSkipField
@total_ordering
class Field(object):
def __init__(self, field_type, resolve=None, null=True, args=None, description='', **extra_args):
SKIP = GraphQLSkipField
creation_counter = 0
def __init__(self, field_type, name=None, resolve=None, required=False, args=None, description='', **extra_args):
self.field_type = field_type
self.resolve_fn = resolve
self.null = null
self.required = required
self.args = args or {}
self.extra_args = extra_args
self._type = None
self.name = name
self.description = description or self.__doc__
self.object_type = None
self.creation_counter = Field.creation_counter
Field.creation_counter += 1
def contribute_to_class(self, cls, name):
if not self.name:
self.name = to_camel_case(name)
self.field_name = name
self.object_type = cls
if isinstance(self.field_type, Field) and not self.field_type.object_type:
self.field_type.contribute_to_class(cls, name)
cls._meta.add_field(self)
def resolver(self, instance, args, info):
if self.object_type.can_resolve(self.field_name, instance, args, info):
return self.resolve(instance, args, info)
else:
return None
def resolve(self, instance, args, info):
if self.resolve_fn:
resolve_fn = self.resolve_fn
else:
resolve_fn = lambda root, args, info: root.resolve(self.field_name, args, info)
resolve_fn = lambda root, args, info: root.resolve(
self.field_name, args, info)
return resolve_fn(instance, args, info)
@cached_property
def type(self):
def get_object_type(self, schema):
field_type = self.field_type
_is_class = inspect.isclass(field_type)
if _is_class and issubclass(field_type, ObjectType):
field_type = field_type._meta.type
elif isinstance(field_type, Field):
field_type = field_type.type
elif field_type == 'self':
field_type = self.object_type._meta.type
field_type = self.type_wrapper(field_type)
return field_type
if isinstance(field_type, Field):
return field_type.get_object_type(schema)
if _is_class and issubclass(field_type, BaseObjectType):
return field_type
elif isinstance(field_type, six.string_types):
if field_type == 'self':
return self.object_type
else:
return schema.get_type(field_type)
def type_wrapper(self, field_type):
if not self.null:
if self.required:
field_type = GraphQLNonNull(field_type)
return field_type
@cached_property
def field(self):
if not self.field_type:
raise Exception('Must specify a field GraphQL type for the field %s'%self.field_name)
@memoize
def internal_type(self, schema):
field_type = self.field_type
if isinstance(field_type, Field):
field_type = self.field_type.internal_type(schema)
else:
object_type = self.get_object_type(schema)
if object_type:
field_type = object_type.internal_type(schema)
field_type = self.type_wrapper(field_type)
return field_type
@memoize
def internal_field(self, schema):
if not self.object_type:
raise Exception('Field could not be constructed in a non graphene.Type or graphene.Interface')
raise Exception(
'Field could not be constructed in a non graphene.Type or graphene.Interface')
extra_args = self.extra_args.copy()
for arg_name, arg_value in extra_args.items():
for arg_name, arg_value in self.extra_args.items():
if isinstance(arg_value, GraphQLArgument):
self.args[arg_name] = arg_value
del extra_args[arg_name]
@ -80,18 +99,22 @@ class Field(object):
raise TypeError("Field %s.%s initiated with invalid args: %s" % (
self.object_type,
self.field_name,
','.join(meta_attrs.keys())
','.join(extra_args.keys())
))
internal_type = self.internal_type(schema)
if not internal_type:
raise Exception("Internal type for field %s is None" % self)
return GraphQLField(
self.type,
internal_type,
description=self.description,
args=self.args,
resolver=self.resolver,
resolver=self.resolve,
)
def __str__(self):
""" Return "object_type.field_name". """
""" Return "object_type.name". """
return '%s.%s' % (self.object_type, self.field_name)
def __repr__(self):
@ -104,8 +127,73 @@ class Field(object):
return '<%s: %s>' % (path, name)
return '<%s>' % path
def __eq__(self, other):
# Needed for @total_ordering
if isinstance(other, Field):
return self.creation_counter == other.creation_counter
return NotImplemented
def __lt__(self, other):
# This is needed because bisect does not take a comparison function.
if isinstance(other, Field):
return self.creation_counter < other.creation_counter
return NotImplemented
def __hash__(self):
return hash(self.creation_counter)
class NativeField(Field):
def __init__(self, field=None):
super(NativeField, self).__init__(None)
self.field = field
def get_field(self, schema):
return self.field
@memoize
def internal_field(self, schema):
return self.get_field(schema)
@memoize
def internal_type(self, schema):
return self.internal_field(schema).type
class LazyField(Field):
@memoize
def inner_field(self, schema):
return self.get_field(schema)
def internal_type(self, schema):
return self.inner_field(schema).internal_type(schema)
def internal_field(self, schema):
return self.inner_field(schema).internal_field(schema)
class LazyNativeField(NativeField):
def __init__(self, *args, **kwargs):
super(LazyNativeField, self).__init__(None, *args, **kwargs)
def get_field(self, schema):
raise NotImplementedError(
"get_field function not implemented for %s LazyField" % self.__class__)
@memoize
def internal_field(self, schema):
return self.get_field(schema)
@memoize
def internal_type(self, schema):
return self.internal_field(schema).type
class TypeField(Field):
def __init__(self, *args, **kwargs):
super(TypeField, self).__init__(self.field_type, *args, **kwargs)
@ -126,6 +214,10 @@ class IDField(TypeField):
field_type = GraphQLID
class FloatField(TypeField):
field_type = GraphQLFloat
class ListField(Field):
def type_wrapper(self, field_type):
return GraphQLList(field_type)

View File

@ -1,8 +1,12 @@
from graphene.utils import cached_property
from collections import OrderedDict
DEFAULT_NAMES = ('description', 'name', 'interface',
'type_name', 'interfaces', 'proxy')
DEFAULT_NAMES = ('description', 'name', 'interface', 'type_name', 'interfaces', 'proxy')
class Options(object):
def __init__(self, meta=None):
self.meta = meta
self.local_fields = []
@ -10,6 +14,7 @@ class Options(object):
self.proxy = False
self.interfaces = []
self.parents = []
self.valid_attrs = DEFAULT_NAMES
def contribute_to_class(self, cls, name):
cls._meta = self
@ -32,7 +37,7 @@ class Options(object):
# over it, so we loop over the *original* dictionary instead.
if name.startswith('_'):
del meta_attrs[name]
for attr_name in DEFAULT_NAMES:
for attr_name in self.valid_attrs:
if attr_name in meta_attrs:
setattr(self, attr_name, meta_attrs.pop(attr_name))
self.original_attrs[attr_name] = getattr(self, attr_name)
@ -40,9 +45,14 @@ class Options(object):
setattr(self, attr_name, getattr(self.meta, attr_name))
self.original_attrs[attr_name] = getattr(self, attr_name)
del self.valid_attrs
# Any leftover attributes must be invalid.
if meta_attrs != {}:
raise TypeError("'class Meta' got invalid attribute(s): %s" % ','.join(meta_attrs.keys()))
raise TypeError(
"'class Meta' got invalid attribute(s): %s" % ','.join(meta_attrs.keys()))
else:
self.proxy = False
if self.interfaces != [] and self.interface:
raise Exception("A interface cannot inherit from interfaces")
@ -51,19 +61,14 @@ class Options(object):
def add_field(self, field):
self.local_fields.append(field)
setattr(self.parent, field.field_name, field)
@cached_property
def fields(self):
fields = []
for parent in self.parents:
fields.extend(parent._meta.fields)
return self.local_fields + fields
return sorted(self.local_fields + fields)
@cached_property
def fields_map(self):
return {f.field_name:f for f in self.fields}
@cached_property
def type(self):
return self.parent.get_graphql_type()
return OrderedDict([(f.field_name, f) for f in self.fields])

10
graphene/core/scalars.py Normal file
View File

@ -0,0 +1,10 @@
from graphql.core.type.definition import GraphQLScalarType
def skip(value):
return None
GraphQLSkipField = GraphQLScalarType(name='SkipField',
serialize=skip,
parse_value=skip,
parse_literal=skip)

124
graphene/core/schema.py Normal file
View File

@ -0,0 +1,124 @@
from functools import wraps
from collections import OrderedDict
from graphql.core import graphql
from graphql.core.type import (
GraphQLSchema as _GraphQLSchema
)
from graphql.core.execution.executor import Executor
from graphql.core.execution.middlewares.sync import SynchronousExecutionMiddleware
from graphql.core.execution import ExecutionResult, execute
from graphql.core.language.parser import parse
from graphql.core.language.source import Source
from graphql.core.validation import validate
from graphql.core.utils.introspection_query import introspection_query
from graphene import signals
from graphene.utils import cached_property
class GraphQLSchema(_GraphQLSchema):
def __init__(self, schema, *args, **kwargs):
self.graphene_schema = schema
super(GraphQLSchema, self).__init__(*args, **kwargs)
class Schema(object):
_query = None
_executor = None
def __init__(self, query=None, mutation=None, name='Schema', executor=None):
self._internal_types = {}
self.mutation = mutation
self.query = query
self.name = name
self.executor = executor
signals.init_schema.send(self)
def __repr__(self):
return '<Schema: %s (%s)>' % (str(self.name), hash(self))
@property
def query(self):
return self._query
@query.setter
def query(self, query):
self._query = query
self._query_type = query and query.internal_type(self)
@property
def executor(self):
if not self._executor:
# TODO: Update to map_type=OrderedDict when graphql-core
# update its package in pypi
self.executor = Executor([SynchronousExecutionMiddleware()])
return self._executor
@executor.setter
def executor(self, value):
self._executor = value
@cached_property
def schema(self):
if not self._query_type:
raise Exception('You have to define a base query type')
return GraphQLSchema(self, query=self._query_type, mutation=self.mutation)
def associate_internal_type(self, internal_type, object_type):
self._internal_types[internal_type.name] = object_type
def register(self, object_type):
self._internal_types[object_type._meta.type_name] = object_type
return object_type
def get_type(self, type_name):
if type_name not in self._internal_types:
raise Exception('Type %s not found in %r' % (type_name, self))
return self._internal_types[type_name]
@property
def types(self):
return self._internal_types
def execute(self, request='', root=None, vars=None, operation_name=None):
root = root or object()
return graphql(
self.schema,
request,
root=self.query(root),
vars=vars,
operation_name=operation_name
)
# source = Source(request, 'GraphQL request')
# ast = parse(source)
# validation_errors = validate(self.schema, ast)
# if validation_errors:
# return ExecutionResult(
# errors=validation_errors,
# invalid=True,
# )
# return self.executor.execute(
# self.schema,
# ast,
# root=self.query(root),
# args=vars,
# operation_name=operation_name,
# validate_ast=False
# )
def introspect(self):
return self.execute(introspection_query).data
def register_internal_type(fun):
@wraps(fun)
def wrapper(cls, schema):
internal_type = fun(cls, schema)
if isinstance(schema, Schema):
schema.associate_internal_type(internal_type, cls)
return internal_type
return wrapper

View File

@ -1,68 +1,98 @@
import inspect
import six
from collections import OrderedDict
from graphql.core.type import (
GraphQLObjectType,
GraphQLInterfaceType,
GraphQLSchema
GraphQLInterfaceType
)
from graphql.core import graphql
from graphene import signals
from graphene.core.options import Options
from graphene.utils import memoize
from graphene.core.schema import register_internal_type
class ObjectTypeMeta(type):
options_cls = Options
def is_interface(cls, parents):
return Interface in parents
def is_mutation(cls, parents):
return Mutation in parents
def __new__(cls, name, bases, attrs):
super_new = super(ObjectTypeMeta, cls).__new__
parents = [b for b in bases if isinstance(b, ObjectTypeMeta)]
parents = [b for b in bases if isinstance(b, cls)]
if not parents:
# If this isn't a subclass of Model, don't do anything special.
return super_new(cls, name, bases, attrs)
module = attrs.pop('__module__')
doc = attrs.pop('__doc__', None)
new_class = super_new(cls, name, bases, {'__module__': module, '__doc__': doc})
new_class = super_new(cls, name, bases, {
'__module__': module,
'__doc__': doc
})
attr_meta = attrs.pop('Meta', None)
if not attr_meta:
meta = getattr(new_class, 'Meta', None)
meta = None
# meta = getattr(new_class, 'Meta', None)
else:
meta = attr_meta
base_meta = getattr(new_class, '_meta', None)
new_class.add_to_class('_meta', Options(meta))
if base_meta and base_meta.proxy:
new_class._meta.interface = base_meta.interface
new_class.add_to_class('_meta', new_class.options_cls(meta))
new_class._meta.interface = new_class.is_interface(parents)
new_class._meta.mutation = new_class.is_mutation(parents)
assert not (new_class._meta.interface and new_class._meta.mutation)
# Add all attributes to the class.
for obj_name, obj in attrs.items():
new_class.add_to_class(obj_name, obj)
new_class.add_extra_fields()
new_fields = new_class._meta.local_fields
field_names = {f.field_name for f in new_fields}
field_names = {f.name: f for f in new_fields}
for base in parents:
original_base = base
if not hasattr(base, '_meta'):
# Things without _meta aren't functional models, so they're
# uninteresting parents.
continue
# if base._meta.schema != new_class._meta.schema:
# raise Exception('The parent schema is not the same')
parent_fields = base._meta.local_fields
# Check for clashes between locally declared fields and those
# on the base classes (we cannot handle shadowed fields at the
# moment).
for field in parent_fields:
if field.field_name in field_names:
raise FieldError(
if field.name in field_names and field.__class__ != field_names[field].__class__:
raise Exception(
'Local field %r in class %r clashes '
'with field of similar name from '
'base class %r' % (field.field_name, name, base.__name__)
'base class %r' % (
field.name, name, base.__name__)
)
new_class._meta.parents.append(base)
if base._meta.interface:
new_class._meta.interfaces.append(base)
# new_class._meta.parents.extend(base._meta.parents)
new_class._prepare()
return new_class
def add_extra_fields(cls):
pass
def _prepare(cls):
signals.class_prepared.send(cls)
def add_to_class(cls, name, value):
# We should call the contribute_to_class method only if it's bound
if not inspect.isclass(value) and hasattr(value, 'contribute_to_class'):
@ -71,66 +101,70 @@ class ObjectTypeMeta(type):
setattr(cls, name, value)
class ObjectType(six.with_metaclass(ObjectTypeMeta)):
def __init__(self, instance=None):
class BaseObjectType(object):
def __new__(cls, instance=None, *args, **kwargs):
if cls._meta.interface:
raise Exception("An interface cannot be initialized")
if instance is None:
return None
elif type(instance) is cls:
instance = instance.instance
return super(BaseObjectType, cls).__new__(cls, *args, **kwargs)
def __init__(self, instance):
signals.pre_init.send(self.__class__, instance=instance)
self.instance = instance
signals.post_init.send(self.__class__, instance=self)
def __getattr__(self, name):
if self.instance:
return getattr(self.instance, name)
def get_field(self, field):
return getattr(self.instance, field, None)
def resolve(self, field_name, args, info):
if field_name not in self._meta.fields_map.keys():
raise Exception('Field %s not found in model'%field_name)
custom_resolve_fn = 'resolve_%s'%field_name
custom_resolve_fn = 'resolve_%s' % field_name
if hasattr(self, custom_resolve_fn):
resolve_fn = getattr(self, custom_resolve_fn)
return resolve_fn(args, info)
return self.get_field(field_name)
@classmethod
def can_resolve(cls, field_name, instance, args, info):
# Useful for manage permissions in fields
return True
def resolve_type(cls, schema, instance, *_):
return instance.internal_type(schema)
@classmethod
def resolve_type(cls, instance, *_):
return instance._meta.type
@classmethod
def get_graphql_type(cls):
fields = cls._meta.fields_map
@memoize
@register_internal_type
def internal_type(cls, schema):
fields_list = cls._meta.fields
fields = lambda: OrderedDict([(f.name, f.internal_field(schema))
for f in fields_list])
if cls._meta.interface:
return GraphQLInterfaceType(
cls._meta.type_name,
description=cls._meta.description,
resolve_type=cls.resolve_type,
fields=lambda: {name:field.field for name, field in fields.items()}
resolve_type=lambda *
args, **kwargs: cls.resolve_type(schema, *args, **kwargs),
fields=fields
)
return GraphQLObjectType(
cls._meta.type_name,
description=cls._meta.description,
interfaces=[i._meta.type for i in cls._meta.interfaces],
fields=lambda: {name:field.field for name, field in fields.items()}
interfaces=[i.internal_type(schema) for i in cls._meta.interfaces],
fields=fields
)
class Interface(ObjectType):
class Meta:
interface = True
proxy = True
class ObjectType(six.with_metaclass(ObjectTypeMeta, BaseObjectType)):
pass
class Schema(object):
def __init__(self, query, mutation=None):
self.query = query
self.query_type = query._meta.type
self._schema = GraphQLSchema(query=self.query_type, mutation=mutation)
def execute(self, request='', root=None, vars=None, operation_name=None):
return graphql(
self._schema,
request=request,
root=root or self.query(),
vars=vars,
operation_name=operation_name
)
class Mutation(six.with_metaclass(ObjectTypeMeta, BaseObjectType)):
pass
class Interface(six.with_metaclass(ObjectTypeMeta, BaseObjectType)):
pass

View File

@ -4,5 +4,5 @@ from functools import wraps
def resolve_only_args(func):
@wraps(func)
def inner(self, args, info):
return func(self, **args)
return inner
return func(self, **args)
return inner

View File

@ -0,0 +1,12 @@
from graphene.relay.fields import (
ConnectionField,
NodeField
)
import graphene.relay.connections
from graphene.relay.types import (
Node
)
from graphene.relay.utils import is_node

View File

@ -0,0 +1,17 @@
from graphql_relay.node.node import (
global_id_field
)
from graphene import signals
from graphene.relay.fields import NodeIDField
from graphene.relay.utils import is_node
@signals.class_prepared.connect
def object_type_created(object_type):
if is_node(object_type):
type_name = object_type._meta.type_name
field = NodeIDField()
object_type.add_to_class('id', field)
assert hasattr(
object_type, 'get_node'), 'get_node classmethod not found in %s Node' % type_name

83
graphene/relay/fields.py Normal file
View File

@ -0,0 +1,83 @@
from collections import Iterable, OrderedDict
from graphql_relay.connection.arrayconnection import (
connection_from_list
)
from graphql_relay.connection.connection import (
connectionArgs
)
from graphql_relay.node.node import (
global_id_field,
from_global_id
)
from graphene.core.fields import Field, LazyNativeField, LazyField
from graphene.utils import cached_property
from graphene.utils import memoize
class ConnectionField(Field):
def __init__(self, field_type, resolve=None, description=''):
super(ConnectionField, self).__init__(field_type, resolve=resolve,
args=connectionArgs, description=description)
def wrap_resolved(self, value, instance, args, info):
return value
def resolve(self, instance, args, info):
resolved = super(ConnectionField, self).resolve(instance, args, info)
if resolved:
resolved = self.wrap_resolved(resolved, instance, args, info)
assert isinstance(
resolved, Iterable), 'Resolved value from the connection field have to be iterable'
return connection_from_list(resolved, args)
@memoize
def internal_type(self, schema):
from graphene.relay.types import BaseNode
object_type = self.get_object_type(schema)
assert issubclass(
object_type, BaseNode), 'Only nodes have connections.'
return object_type.get_connection(schema)
class NodeField(LazyNativeField):
def __init__(self, object_type=None, *args, **kwargs):
super(NodeField, self).__init__(*args, **kwargs)
self.field_object_type = object_type
def get_field(self, schema):
if self.field_object_type:
field = NodeTypeField(self.field_object_type)
field.contribute_to_class(self.object_type, self.field_name)
return field.internal_field(schema)
from graphene.relay.types import BaseNode
return BaseNode.get_definitions(schema).node_field
class NodeTypeField(LazyField):
def __init__(self, object_type, *args, **kwargs):
super(NodeTypeField, self).__init__(None, *args, **kwargs)
self.field_object_type = object_type
def inner_field(self, schema):
from graphene.relay.types import BaseNode
node_field = BaseNode.get_definitions(schema).node_field
def resolver(instance, args, info):
global_id = args.get('id')
resolved_global_id = from_global_id(global_id)
if resolved_global_id.type == self.field_object_type._meta.type_name:
return node_field.resolver(instance, args, info)
args = OrderedDict([(a.name, a) for a in node_field.args])
field = Field(self.field_object_type, id=args['id'], resolve=resolver)
field.contribute_to_class(self.object_type, self.field_name)
return field
class NodeIDField(LazyNativeField):
def get_field(self, schema):
return global_id_field(self.object_type._meta.type_name)

52
graphene/relay/types.py Normal file
View File

@ -0,0 +1,52 @@
from graphql_relay.node.node import (
node_definitions,
from_global_id
)
from graphql_relay.connection.connection import (
connection_definitions
)
from graphene.core.types import Interface
from graphene.core.fields import LazyNativeField
from graphene.utils import memoize
def get_node_type(schema, obj):
return obj.internal_type(schema)
def get_node(schema, global_id, *args):
resolved_global_id = from_global_id(global_id)
_type, _id = resolved_global_id.type, resolved_global_id.id
object_type = schema.get_type(_type)
if not object_type or not issubclass(object_type, BaseNode):
raise Exception("The type %s is not a Node" % _type)
return object_type.get_node(_id)
class BaseNode(object):
@classmethod
@memoize
def get_definitions(cls, schema):
return node_definitions(lambda *args: get_node(schema, *args), lambda *args: get_node_type(schema, *args))
@classmethod
@memoize
def get_connection(cls, schema):
_type = cls.internal_type(schema)
type_name = cls._meta.type_name
connection = connection_definitions(type_name, _type).connection_type
return connection
@classmethod
def internal_type(cls, schema):
from graphene.relay.utils import is_node_type
if is_node_type(cls):
# Return only node_interface when is the Node Inerface
return BaseNode.get_definitions(schema).node_interface
return super(BaseNode, cls).internal_type(schema)
class Node(BaseNode, Interface):
pass

9
graphene/relay/utils.py Normal file
View File

@ -0,0 +1,9 @@
from graphene.relay.types import BaseNode
def is_node(object_type):
return issubclass(object_type, BaseNode) and not is_node_type(object_type)
def is_node_type(object_type):
return BaseNode in object_type.__bases__

6
graphene/signals.py Normal file
View File

@ -0,0 +1,6 @@
from blinker import Signal
init_schema = Signal()
class_prepared = Signal()
pre_init = Signal()
post_init = Signal()

View File

@ -1,3 +1,6 @@
from functools import wraps
class cached_property(object):
"""
A property that is only computed once per instance and then replaces itself
@ -14,3 +17,70 @@ class cached_property(object):
return self
value = obj.__dict__[self.func.__name__] = self.func(obj)
return value
def memoize(fun):
"""A simple memoize decorator for functions supporting positional args."""
@wraps(fun)
def wrapper(*args, **kwargs):
key = (args, frozenset(sorted(kwargs.items())))
try:
return cache[key]
except KeyError:
ret = cache[key] = fun(*args, **kwargs)
return ret
cache = {}
return wrapper
# From this response in Stackoverflow
# http://stackoverflow.com/a/19053800/1072990
def to_camel_case(snake_str):
components = snake_str.split('_')
# We capitalize the first letter of each component except the first one
# with the 'title' method and join them together.
return components[0] + "".join(x.title() for x in components[1:])
class LazyMap(object):
def __init__(self, origin, _map, state=None):
self._origin = origin
self._origin_iter = origin.__iter__()
self._state = state or []
self._finished = False
self._map = _map
def __iter__(self):
return self if not self._finished else iter(self._state)
def iter(self):
return self.__iter__()
def __len__(self):
return self._origin.__len__()
def __next__(self):
try:
n = next(self._origin_iter)
n = self._map(n)
except StopIteration as e:
self._finished = True
raise e
else:
self._state.append(n)
return n
def next(self):
return self.__next__()
def __getitem__(self, key):
item = self._origin.__getitem__(key)
if isinstance(key, slice):
return LazyMap(item, self._map)
return self._map(item)
def __getattr__(self, name):
return getattr(self._origin, name)
def __repr__(self):
return "<LazyMap %s>" % repr(self._origin)

View File

@ -17,18 +17,19 @@ class PyTest(TestCommand):
self.test_suite = True
def run_tests(self):
#import here, cause outside the eggs aren't loaded
# import here, cause outside the eggs aren't loaded
import pytest
errno = pytest.main(self.pytest_args)
sys.exit(errno)
setup(
name='graphene',
version='0.1',
version='0.1.4',
description='Graphene: GraphQL Object Mapper',
description='Graphene: Python DSL for GraphQL',
long_description=open('README.rst').read(),
url='https://github.com/syrusakbary/graphene',
url='https://github.com/graphql-python/graphene',
author='Syrus Akbary',
author_email='me@syrusakbary.com',
@ -40,6 +41,12 @@ setup(
'Intended Audience :: Developers',
'Topic :: Software Development :: Libraries',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: Implementation :: PyPy',
],
keywords='api graphql protocol rest relay graphene',
@ -47,14 +54,18 @@ setup(
packages=find_packages(exclude=['tests']),
install_requires=[
'six',
'graphqllib',
'graphql-relay'
'six>=1.10.0',
'blinker',
'graphql-core==0.1a2',
'graphql-relay==0.2.0'
],
tests_require=[
'pytest>=2.7.2',
'pytest-django',
],
tests_require=['pytest>=2.7.2'],
extras_require={
'django': [
'Django>=1.8.0,<1.9',
'Django>=1.6.0,<1.9',
'singledispatch>=3.4.0.3',
],
},

0
tests/__init__.py Normal file
View File

View File

View File

@ -0,0 +1,19 @@
from datetime import date
from .models import Reporter, Article
r = Reporter(first_name='John', last_name='Smith', email='john@example.com')
r.save()
r2 = Reporter(first_name='Paul', last_name='Jones', email='paul@example.com')
r2.save()
a = Article(id=None, headline="This is a test",
pub_date=date(2005, 7, 27), reporter=r)
a.save()
new_article = r.articles.create(
headline="John's second story", pub_date=date(2005, 7, 29))
new_article2 = Article(headline="Paul's story", pub_date=date(2006, 1, 17))
r.articles.add(new_article2)

View File

@ -0,0 +1,32 @@
from __future__ import absolute_import
from django.db import models
class Pet(models.Model):
name = models.CharField(max_length=30)
class Reporter(models.Model):
first_name = models.CharField(max_length=30)
last_name = models.CharField(max_length=30)
email = models.EmailField()
pets = models.ManyToManyField('self')
def __str__(self): # __unicode__ on Python 2
return "%s %s" % (self.first_name, self.last_name)
class Meta:
app_label = 'contrib_django'
class Article(models.Model):
headline = models.CharField(max_length=100)
pub_date = models.DateField()
reporter = models.ForeignKey(Reporter, related_name='articles')
def __str__(self): # __unicode__ on Python 2
return self.headline
class Meta:
ordering = ('headline',)
app_label = 'contrib_django'

View File

@ -0,0 +1,114 @@
from py.test import raises
from collections import namedtuple
from pytest import raises
import graphene
from graphene import relay
from graphene.contrib.django.converter import (
convert_django_field
)
from graphene.contrib.django.fields import (
ConnectionOrListField,
DjangoModelField
)
from django.db import models
from .models import Article, Reporter
def assert_conversion(django_field, graphene_field, *args):
field = django_field(*args, help_text='Custom Help Text')
graphene_type = convert_django_field(field)
assert isinstance(graphene_type, graphene_field)
assert graphene_type.description == 'Custom Help Text'
return graphene_type
def test_should_unknown_django_field_raise_exception():
with raises(Exception) as excinfo:
convert_django_field(None)
assert 'Don\'t know how to convert the Django field' in str(excinfo.value)
def test_should_date_convert_string():
assert_conversion(models.DateField, graphene.StringField)
def test_should_char_convert_string():
assert_conversion(models.CharField, graphene.StringField)
def test_should_text_convert_string():
assert_conversion(models.TextField, graphene.StringField)
def test_should_email_convert_string():
assert_conversion(models.EmailField, graphene.StringField)
def test_should_slug_convert_string():
assert_conversion(models.SlugField, graphene.StringField)
def test_should_url_convert_string():
assert_conversion(models.URLField, graphene.StringField)
def test_should_auto_convert_id():
assert_conversion(models.AutoField, graphene.IDField)
def test_should_positive_integer_convert_int():
assert_conversion(models.PositiveIntegerField, graphene.IntField)
def test_should_positive_small_convert_int():
assert_conversion(models.PositiveSmallIntegerField, graphene.IntField)
def test_should_small_integer_convert_int():
assert_conversion(models.SmallIntegerField, graphene.IntField)
def test_should_big_integer_convert_int():
assert_conversion(models.BigIntegerField, graphene.IntField)
def test_should_integer_convert_int():
assert_conversion(models.IntegerField, graphene.IntField)
def test_should_boolean_convert_boolean():
field = assert_conversion(models.BooleanField, graphene.BooleanField)
assert field.required is True
def test_should_nullboolean_convert_boolean():
field = assert_conversion(models.NullBooleanField, graphene.BooleanField)
assert field.required is False
def test_should_float_convert_float():
assert_conversion(models.FloatField, graphene.FloatField)
def test_should_manytomany_convert_connectionorlist():
graphene_type = convert_django_field(Reporter._meta.local_many_to_many[0])
assert isinstance(graphene_type, ConnectionOrListField)
assert isinstance(graphene_type.field_type, DjangoModelField)
assert graphene_type.field_type.model == Reporter
def test_should_manytoone_convert_connectionorlist():
graphene_type = convert_django_field(Reporter.articles.related)
assert isinstance(graphene_type, ConnectionOrListField)
assert isinstance(graphene_type.field_type, DjangoModelField)
assert graphene_type.field_type.model == Article
def test_should_onetoone_convert_model():
field = assert_conversion(models.OneToOneField, DjangoModelField, Article)
assert field.model == Article
def test_should_foreignkey_convert_model():
field = assert_conversion(models.ForeignKey, DjangoModelField, Article)
assert field.model == Article

View File

@ -0,0 +1,184 @@
from py.test import raises
from collections import namedtuple
from pytest import raises
import graphene
from graphene import relay
from graphene.contrib.django import (
DjangoObjectType,
DjangoNode
)
from .models import Reporter, Article
from tests.utils import assert_equal_lists
def test_should_raise_if_no_model():
with raises(Exception) as excinfo:
class Character1(DjangoObjectType):
pass
assert 'model in the Meta' in str(excinfo.value)
def test_should_raise_if_model_is_invalid():
with raises(Exception) as excinfo:
class Character2(DjangoObjectType):
class Meta:
model = 1
assert 'not a Django model' in str(excinfo.value)
def test_should_raise_if_model_is_invalid():
with raises(Exception) as excinfo:
class ReporterTypeError(DjangoObjectType):
class Meta:
model = Reporter
only_fields = ('articles', )
schema = graphene.Schema(query=ReporterTypeError)
query = '''
query ReporterQuery {
articles
}
'''
result = schema.execute(query)
assert not result.errors
def test_should_map_fields_correctly():
class ReporterType2(DjangoObjectType):
class Meta:
model = Reporter
assert_equal_lists(
ReporterType2._meta.fields_map.keys(),
['articles', 'first_name', 'last_name', 'email', 'pets', 'id']
)
def test_should_map_fields():
class ReporterType(DjangoObjectType):
class Meta:
model = Reporter
class Query2(graphene.ObjectType):
reporter = graphene.Field(ReporterType)
def resolve_reporter(self, *args, **kwargs):
return ReporterType(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=Query2)
result = Schema.execute(query)
assert not result.errors
assert result.data == expected
def test_should_map_only_few_fields():
class Reporter2(DjangoObjectType):
class Meta:
model = Reporter
only_fields = ('id', 'email')
assert_equal_lists(
Reporter2._meta.fields_map.keys(),
['id', 'email']
)
def test_should_node():
class ReporterNodeType(DjangoNode):
class Meta:
model = Reporter
@classmethod
def get_node(cls, id):
return ReporterNodeType(Reporter(id=2, first_name='Cookie Monster'))
def resolve_articles(self, *args, **kwargs):
return [ArticleNodeType(Article(headline='Hi!'))]
class ArticleNodeType(DjangoNode):
class Meta:
model = Article
@classmethod
def get_node(cls, id):
return ArticleNodeType(Article(id=1, headline='Article node'))
class Query1(graphene.ObjectType):
node = relay.NodeField()
reporter = graphene.Field(ReporterNodeType)
article = graphene.Field(ArticleNodeType)
def resolve_reporter(self, *args, **kwargs):
return ReporterNodeType(Reporter(id=1, first_name='ABA', last_name='X'))
query = '''
query ReporterQuery {
reporter {
id,
firstName,
articles {
edges {
node {
headline
}
}
}
lastName,
email
}
myArticle: node(id:"QXJ0aWNsZU5vZGVUeXBlOjE=") {
id
... on ReporterNodeType {
firstName
}
... on ArticleNodeType {
headline
}
}
}
'''
expected = {
'reporter': {
'id': 'UmVwb3J0ZXJOb2RlVHlwZTox',
'firstName': 'ABA',
'lastName': 'X',
'email': '',
'articles': {
'edges': [{
'node': {
'headline': 'Hi!'
}
}]
},
},
'myArticle': {
'id': 'QXJ0aWNsZU5vZGVUeXBlOjE=',
'headline': 'Article node'
}
}
Schema = graphene.Schema(query=Query1)
result = Schema.execute(query)
assert not result.errors
assert result.data == expected

View File

@ -0,0 +1,75 @@
from py.test import raises
from collections import namedtuple
from pytest import raises
from graphene.core.fields import (
Field,
StringField,
)
from graphql.core.type import (
GraphQLObjectType,
GraphQLInterfaceType
)
from graphene import Schema
from graphene.contrib.django.types import (
DjangoNode,
DjangoInterface
)
from .models import Reporter, Article
from tests.utils import assert_equal_lists
class Character(DjangoInterface):
'''Character description'''
class Meta:
model = Reporter
class Human(DjangoNode):
'''Human description'''
def get_node(self, id):
pass
class Meta:
model = Article
schema = Schema()
def test_django_interface():
assert DjangoNode._meta.interface is True
def test_pseudo_interface():
object_type = Character.internal_type(schema)
assert Character._meta.interface is True
assert isinstance(object_type, GraphQLInterfaceType)
assert Character._meta.model == Reporter
assert_equal_lists(
object_type.get_fields().keys(),
['articles', 'firstName', 'lastName', 'email', 'pets', 'id']
)
def test_interface_resolve_type():
resolve_type = Character.resolve_type(schema, Human(object()))
assert isinstance(resolve_type, GraphQLObjectType)
def test_object_type():
object_type = Human.internal_type(schema)
fields_map = Human._meta.fields_map
assert Human._meta.interface is False
assert isinstance(object_type, GraphQLObjectType)
assert object_type.get_fields() == {
'headline': fields_map['headline'].internal_field(schema),
'id': fields_map['id'].internal_field(schema),
'reporter': fields_map['reporter'].internal_field(schema),
'pubDate': fields_map['pub_date'].internal_field(schema),
}
assert object_type.get_interfaces() == [DjangoNode.internal_type(schema)]

View File

@ -0,0 +1,40 @@
from django.conf.urls import url
from graphene.contrib.django.views import GraphQLView
import graphene
from graphene import Schema
from graphene.contrib.django.types import (
DjangoNode,
DjangoInterface
)
from .models import Reporter, Article
class Character(DjangoNode):
class Meta:
model = Reporter
def get_node(self, id):
pass
class Human(DjangoNode):
raises = graphene.StringField()
class Meta:
model = Article
def resolve_raises(self, *args):
raise Exception("This field should raise exception")
def get_node(self, id):
pass
schema = Schema(query=Human)
urlpatterns = [
url(r'^graphql', GraphQLView.as_view(schema=schema)),
]

View File

@ -0,0 +1,101 @@
import json
import pytest
from py.test import raises
from collections import namedtuple
from pytest import raises
from graphene.core.fields import (
Field,
StringField,
)
from graphql.core.type import (
GraphQLObjectType,
GraphQLInterfaceType
)
from graphene import Schema
from graphene.contrib.django.types import (
DjangoNode,
DjangoInterface
)
def format_response(response):
return json.loads(response.content.decode())
def test_client_get_no_query(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.get('/graphql')
json_response = format_response(response)
assert json_response == {'errors': [{'message': 'Must provide query string.'}]}
def test_client_post_no_query(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.post('/graphql', {})
json_response = format_response(response)
assert json_response == {'errors': [{'message': 'Must provide query string.'}]}
def test_client_post_malformed_json(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.post('/graphql', 'MALFORMED', 'application/json')
json_response = format_response(response)
assert json_response == {'errors': [{'message': 'Malformed json body in the post data'}]}
def test_client_post_empty_query(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.post('/graphql', json.dumps({'query': ''}), 'application/json')
json_response = format_response(response)
assert json_response == {'errors': [{'message': 'Must provide query string.'}]}
def test_client_post_bad_query(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.post('/graphql', json.dumps({'query': '{ MALFORMED'}), 'application/json')
json_response = format_response(response)
assert 'errors' in json_response
assert len(json_response['errors']) == 1
assert 'Syntax Error GraphQL' in json_response['errors'][0]['message']
def test_client_get_good_query(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.get('/graphql', {'query': '{ headline }'})
json_response = format_response(response)
expected_json = {
'data': {
'headline': None
}
}
assert json_response == expected_json
def test_client_get_good_query_with_raise(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.get('/graphql', {'query': '{ raises }'})
json_response = format_response(response)
assert json_response['errors'][0]['message'] == 'This field should raise exception'
assert json_response['data']['raises'] is None
def test_client_post_good_query(settings, client):
settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
response = client.post('/graphql', json.dumps({'query': '{ headline }'}), 'application/json')
json_response = format_response(response)
expected_json = {
'data': {
'headline': None
}
}
assert json_response == expected_json
# def test_client_get_bad_query(settings, client):
# settings.ROOT_URLCONF = 'tests.contrib_django.test_urls'
# response = client.get('/graphql')
# json_response = format_response(response)
# assert json_response == {'errors': [{'message': 'Must provide query string.'}]}

View File

@ -4,6 +4,7 @@ from pytest import raises
from graphene.core.fields import (
Field,
StringField,
NonNullField
)
from graphene.core.options import Options
@ -17,45 +18,161 @@ from graphql.core.type import (
GraphQLID,
)
class ObjectType(object):
_meta = Options()
def resolve(self, *args, **kwargs):
return None
def can_resolve(self, *args):
return True
def __str__(self):
return "ObjectType"
ot = ObjectType()
ObjectType._meta.contribute_to_class(ObjectType, '_meta')
class Schema(object):
pass
schema = Schema()
def test_field_no_contributed_raises_error():
f = Field(GraphQLString)
with raises(Exception) as excinfo:
f.field
f.internal_field(schema)
def test_field_type():
f = Field(GraphQLString)
f.contribute_to_class(ot, 'field_name')
assert isinstance(f.field, GraphQLField)
assert f.type == GraphQLString
assert isinstance(f.internal_field(schema), GraphQLField)
assert f.internal_type(schema) == GraphQLString
def test_field_name_automatic_camelcase():
f = Field(GraphQLString)
f.contribute_to_class(ot, 'field_name')
assert f.name == 'fieldName'
def test_field_name_use_name_if_exists():
f = Field(GraphQLString, name='my_custom_name')
f.contribute_to_class(ot, 'field_name')
assert f.name == 'my_custom_name'
def test_stringfield_type():
f = StringField()
f.contribute_to_class(ot, 'field_name')
assert f.type == GraphQLString
assert f.internal_type(schema) == GraphQLString
def test_stringfield_type_null():
f = StringField(null=False)
def test_nonnullfield_type():
f = NonNullField(StringField())
f.contribute_to_class(ot, 'field_name')
assert isinstance(f.field, GraphQLField)
assert isinstance(f.type, GraphQLNonNull)
assert isinstance(f.internal_type(schema), GraphQLNonNull)
def test_stringfield_type_required():
f = StringField(required=True)
f.contribute_to_class(ot, 'field_name')
assert isinstance(f.internal_field(schema), GraphQLField)
assert isinstance(f.internal_type(schema), GraphQLNonNull)
def test_field_resolve():
f = StringField(null=False)
f = StringField(required=True, resolve=lambda *args: 'RESOLVED')
f.contribute_to_class(ot, 'field_name')
field_type = f.field
field_type.resolver(ot,2,3)
field_type = f.internal_field(schema)
assert 'RESOLVED' == field_type.resolver(ot, 2, 3)
def test_field_resolve_type_custom():
class MyCustomType(object):
pass
class Schema(object):
def get_type(self, name):
if name == 'MyCustomType':
return MyCustomType
s = Schema()
f = Field('MyCustomType')
f.contribute_to_class(ot, 'field_name')
field_type = f.get_object_type(s)
assert field_type == MyCustomType
def test_field_resolve_type_custom():
s = Schema()
f = Field('self')
f.contribute_to_class(ot, 'field_name')
field_type = f.get_object_type(s)
assert field_type == ot
def test_field_orders():
f1 = Field(None)
f2 = Field(None)
assert f1 < f2
def test_field_orders_wrong_type():
field = Field(None)
try:
assert not field < 1
except TypeError:
# Fix exception raising in Python3+
pass
def test_field_eq():
f1 = Field(None)
f2 = Field(None)
assert f1 != f2
def test_field_eq_wrong_type():
field = Field(None)
assert field != 1
def test_field_hash():
f1 = Field(None)
f2 = Field(None)
assert hash(f1) != hash(f2)
def test_field_none_type_raises_error():
s = Schema()
f = Field(None)
f.contribute_to_class(ot, 'field_name')
with raises(Exception) as excinfo:
f.internal_field(s)
assert str(excinfo.value) == "Internal type for field ObjectType.field_name is None"
def test_field_str():
f = StringField()
f.contribute_to_class(ot, 'field_name')
assert str(f) == "ObjectType.field_name"
def test_field_repr():
f = StringField()
assert repr(f) == "<graphene.core.fields.StringField>"
def test_field_repr_contributed():
f = StringField()
f.contribute_to_class(ot, 'field_name')
assert repr(f) == "<graphene.core.fields.StringField: field_name>"

View File

@ -8,13 +8,16 @@ from graphene.core.fields import (
from graphene.core.options import Options
class Meta:
interface = True
type_name = 'Character'
class InvalidMeta:
other_value = True
def test_field_added_in_meta():
opt = Options(Meta)
@ -27,6 +30,7 @@ def test_field_added_in_meta():
opt.add_field(f)
assert f in opt.fields
def test_options_contribute():
opt = Options(Meta)
@ -36,6 +40,7 @@ def test_options_contribute():
opt.contribute_to_class(ObjectType, '_meta')
assert ObjectType._meta == opt
def test_options_typename():
opt = Options(Meta)
@ -45,16 +50,19 @@ def test_options_typename():
opt.contribute_to_class(ObjectType, '_meta')
assert opt.type_name == 'Character'
def test_options_description():
opt = Options(Meta)
class ObjectType(object):
'''False description'''
pass
opt.contribute_to_class(ObjectType, '_meta')
assert opt.description == 'False description'
def test_field_no_contributed_raises_error():
opt = Options(InvalidMeta)

68
tests/core/test_query.py Normal file
View File

@ -0,0 +1,68 @@
from py.test import raises
from collections import namedtuple
from pytest import raises
from graphql.core import graphql
from graphene.core.fields import (
Field,
StringField,
ListField,
)
from graphql.core.type import (
GraphQLObjectType,
GraphQLSchema,
GraphQLInterfaceType
)
from graphene.core.types import (
Interface,
ObjectType
)
class Character(Interface):
name = StringField()
class Pet(ObjectType):
type = StringField(resolve=lambda *_: 'Dog')
class Human(Character):
friends = ListField(Character)
pet = Field(Pet)
def resolve_name(self, *args):
return 'Peter'
def resolve_friend(self, *args):
return Human(object())
def resolve_pet(self, *args):
return Pet(object())
# def resolve_friends(self, *args, **kwargs):
# return 'HEY YOU!'
schema = object()
Human_type = Human.internal_type(schema)
def test_query():
schema = GraphQLSchema(query=Human_type)
query = '''
{
name
pet {
type
}
}
'''
expected = {
'name': 'Peter',
'pet': {
'type': 'Dog'
}
}
result = graphql(schema, query, root=Human(object()))
assert not result.errors
assert result.data == expected

View File

@ -0,0 +1,18 @@
from graphene.core.scalars import (
GraphQLSkipField
)
def test_skipfield_serialize():
f = GraphQLSkipField
assert f.serialize('a') is None
def test_skipfield_parse_value():
f = GraphQLSkipField
assert f.parse_value('a') is None
def test_skipfield_parse_literal():
f = GraphQLSkipField
assert f.parse_literal('a') is None

141
tests/core/test_schema.py Normal file
View File

@ -0,0 +1,141 @@
from py.test import raises
from collections import namedtuple
from pytest import raises
from graphql.core import graphql
from graphene.core.fields import (
Field,
StringField,
ListField,
)
from graphql.core.type import (
GraphQLObjectType,
GraphQLSchema,
GraphQLInterfaceType
)
from graphene import (
Interface,
ObjectType,
Schema
)
from tests.utils import assert_equal_lists
schema = Schema(name='My own schema')
class Character(Interface):
name = StringField()
class Pet(ObjectType):
type = StringField(resolve=lambda *_: 'Dog')
class Human(Character):
friends = ListField(Character)
pet = Field(Pet)
def resolve_name(self, *args):
return 'Peter'
def resolve_friend(self, *args):
return Human(object())
def resolve_pet(self, *args):
return Pet(object())
schema.query = Human
def test_get_registered_type():
assert schema.get_type('Character') == Character
def test_get_unregistered_type():
with raises(Exception) as excinfo:
schema.get_type('NON_EXISTENT_MODEL')
assert 'not found' in str(excinfo.value)
def test_schema_query():
assert schema.query == Human
def test_query_schema_graphql():
a = object()
query = '''
{
name
pet {
type
}
}
'''
expected = {
'name': 'Peter',
'pet': {
'type': 'Dog'
}
}
result = graphql(schema.schema, query, root=Human(object()))
assert not result.errors
assert result.data == expected
def test_query_schema_execute():
a = object()
query = '''
{
name
pet {
type
}
}
'''
expected = {
'name': 'Peter',
'pet': {
'type': 'Dog'
}
}
result = schema.execute(query, root=object())
assert not result.errors
assert result.data == expected
def test_schema_get_type_map():
assert_equal_lists(
schema.schema.get_type_map().keys(),
['__Field', 'String', 'Pet', 'Character', '__InputValue', '__Directive', '__TypeKind', '__Schema', '__Type', 'Human', '__EnumValue', 'Boolean']
)
def test_schema_no_query():
schema = Schema(name='My own schema')
with raises(Exception) as excinfo:
schema.schema
assert 'define a base query type' in str(excinfo)
def test_schema_register():
schema = Schema(name='My own schema')
@schema.register
class MyType(ObjectType):
type = StringField(resolve=lambda *_: 'Dog')
assert schema.get_type('MyType') == MyType
def test_schema_introspect():
schema = Schema(name='My own schema')
class MyType(ObjectType):
type = StringField(resolve=lambda *_: 'Dog')
schema.query = MyType
introspection = schema.introspect()
assert '__schema' in introspection

View File

@ -15,27 +15,48 @@ from graphene.core.types import (
ObjectType
)
class Character(Interface):
'''Character description'''
name = StringField()
class Meta:
type_name = 'core.Character'
class Human(Character):
'''Human description'''
friends = StringField()
class Meta:
type_name = 'core.Human'
schema = object()
def test_interface():
object_type = Character._meta.type
object_type = Character.internal_type(schema)
assert Character._meta.interface == True
assert Character._meta.type_name == 'Character'
assert isinstance(object_type, GraphQLInterfaceType)
assert Character._meta.type_name == 'core.Character'
assert object_type.description == 'Character description'
assert object_type.get_fields() == {'name': Character.name.field}
assert object_type.get_fields() == {
'name': Character._meta.fields_map['name'].internal_field(schema)}
def test_interface_resolve_type():
resolve_type = Character.resolve_type(schema, Human(object()))
assert isinstance(resolve_type, GraphQLObjectType)
def test_object_type():
object_type = Human._meta.type
object_type = Human.internal_type(schema)
assert Human._meta.interface == False
assert Human._meta.type_name == 'Human'
assert Human._meta.type_name == 'core.Human'
assert isinstance(object_type, GraphQLObjectType)
assert object_type.description == 'Human description'
assert object_type.get_fields() == {'name': Character.name.field, 'friends': Human.friends.field}
assert object_type.get_interfaces() == [Character._meta.type]
assert object_type.get_fields() == {'name': Character._meta.fields_map['name'].internal_field(
schema), 'friends': Human._meta.fields_map['friends'].internal_field(schema)}
assert object_type.get_interfaces() == [Character.internal_type(schema)]

14
tests/django_settings.py Normal file
View File

@ -0,0 +1,14 @@
SECRET_KEY = 1
INSTALLED_APPS = [
'graphene.contrib.django',
'tests.starwars_django',
'tests.contrib_django',
]
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': 'tests/django.sqlite',
}
}

52
tests/relay/test_relay.py Normal file
View File

@ -0,0 +1,52 @@
from pytest import raises
import graphene
from graphene import relay
schema = graphene.Schema()
class OtherNode(relay.Node):
name = graphene.StringField()
@classmethod
def get_node(cls, id):
pass
def test_field_no_contributed_raises_error():
with raises(Exception) as excinfo:
class Part(relay.Node):
x = graphene.StringField()
assert 'get_node' in str(excinfo.value)
def test_node_should_have_same_connection_always():
s = object()
connection1 = OtherNode.get_connection(s)
connection2 = OtherNode.get_connection(s)
assert connection1 == connection2
def test_node_should_have_id_field():
assert 'id' in OtherNode._meta.fields_map
# def test_field_no_contributed_raises_error():
# with raises(Exception) as excinfo:
# class Ship(graphene.ObjectType):
# name = graphene.StringField()
# class Meta:
# schema = schema
# class Faction(relay.Node):
# name = graphene.StringField()
# ships = relay.ConnectionField(Ship)
# @classmethod
# def get_node(cls):
# pass
# class Meta:
# schema = schema
# assert 'same type_name' in str(excinfo.value)

View File

@ -0,0 +1,43 @@
from pytest import raises
import graphene
from graphene import relay
schema = graphene.Schema()
class MyType(object):
name = 'my'
class MyNode(relay.Node):
name = graphene.StringField()
@classmethod
def get_node(cls, id):
return MyNode(MyType())
class Query(graphene.ObjectType):
my_node = relay.NodeField(MyNode)
schema.query = Query
def test_nodefield_query():
query = '''
query RebelsShipsQuery {
myNode(id:"TXlOb2RlOjE=") {
name
}
}
'''
expected = {
'myNode': {
'name': 'my'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected

View File

@ -5,40 +5,40 @@ Human = namedtuple('Human', 'id name friends appearsIn homePlanet')
luke = Human(
id='1000',
name='Luke Skywalker',
friends=[ '1002', '1003', '2000', '2001' ],
appearsIn=[ 4, 5, 6 ],
friends=['1002', '1003', '2000', '2001'],
appearsIn=[4, 5, 6],
homePlanet='Tatooine',
)
vader = Human(
id='1001',
name='Darth Vader',
friends=[ '1004' ],
appearsIn=[ 4, 5, 6 ],
friends=['1004'],
appearsIn=[4, 5, 6],
homePlanet='Tatooine',
)
han = Human(
id='1002',
name='Han Solo',
friends=[ '1000', '1003', '2001' ],
appearsIn=[ 4, 5, 6 ],
friends=['1000', '1003', '2001'],
appearsIn=[4, 5, 6],
homePlanet=None,
)
leia = Human(
id='1003',
name='Leia Organa',
friends=[ '1000', '1002', '2000', '2001' ],
appearsIn=[ 4, 5, 6 ],
friends=['1000', '1002', '2000', '2001'],
appearsIn=[4, 5, 6],
homePlanet='Alderaan',
)
tarkin = Human(
id='1004',
name='Wilhuff Tarkin',
friends=[ '1001' ],
appearsIn=[ 4 ],
friends=['1001'],
appearsIn=[4],
homePlanet=None,
)
@ -55,16 +55,16 @@ Droid = namedtuple('Droid', 'id name friends appearsIn primaryFunction')
threepio = Droid(
id='2000',
name='C-3PO',
friends=[ '1000', '1002', '1003', '2001' ],
appearsIn=[ 4, 5, 6 ],
friends=['1000', '1002', '1003', '2001'],
appearsIn=[4, 5, 6],
primaryFunction='Protocol',
)
artoo = Droid(
id='2001',
name='R2-D2',
friends=[ '1000', '1002', '1003' ],
appearsIn=[ 4, 5, 6 ],
friends=['1000', '1002', '1003'],
appearsIn=[4, 5, 6],
primaryFunction='Astromech',
)
@ -73,6 +73,7 @@ droidData = {
'2001': artoo,
}
def getCharacter(id):
return humanData.get(id) or droidData.get(id)

View File

@ -4,17 +4,19 @@ from graphene import resolve_only_args
from .data import getHero, getHuman, getCharacter, getDroid, Human as _Human, Droid as _Droid
Episode = graphene.Enum('Episode', dict(
NEWHOPE = 4,
EMPIRE = 5,
JEDI = 6
NEWHOPE=4,
EMPIRE=5,
JEDI=6
))
def wrap_character(character):
if isinstance(character, _Human):
if isinstance(character, _Human):
return Human(character)
elif isinstance(character, _Droid):
return Droid(character)
class Character(graphene.Interface):
id = graphene.IDField()
name = graphene.StringField()
@ -24,6 +26,7 @@ class Character(graphene.Interface):
def resolve_friends(self, args, *_):
return [wrap_character(getCharacter(f)) for f in self.instance.friends]
class Human(Character):
homePlanet = graphene.StringField()
@ -34,24 +37,25 @@ class Droid(Character):
class Query(graphene.ObjectType):
hero = graphene.Field(Character,
episode = graphene.Argument(Episode)
)
episode=graphene.Argument(Episode)
)
human = graphene.Field(Human,
id = graphene.Argument(graphene.String)
)
id=graphene.Argument(graphene.String)
)
droid = graphene.Field(Droid,
id = graphene.Argument(graphene.String)
)
id=graphene.Argument(graphene.String)
)
class Meta:
type_name = 'core.Query'
@resolve_only_args
def resolve_hero(self, episode):
def resolve_hero(self, episode=None):
return wrap_character(getHero(episode))
@resolve_only_args
def resolve_human(self, id):
return wrap_character(getHuman(id))
if human:
return Human(human)
@resolve_only_args
def resolve_droid(self, id):

View File

@ -69,7 +69,7 @@ def test_nested_query():
'friends': [
{
'name': 'Luke Skywalker',
'appearsIn': [ 'NEWHOPE', 'EMPIRE', 'JEDI' ],
'appearsIn': ['NEWHOPE', 'EMPIRE', 'JEDI'],
'friends': [
{
'name': 'Han Solo',
@ -87,7 +87,7 @@ def test_nested_query():
},
{
'name': 'Han Solo',
'appearsIn': [ 'NEWHOPE', 'EMPIRE', 'JEDI' ],
'appearsIn': ['NEWHOPE', 'EMPIRE', 'JEDI'],
'friends': [
{
'name': 'Luke Skywalker',
@ -102,7 +102,7 @@ def test_nested_query():
},
{
'name': 'Leia Organa',
'appearsIn': [ 'NEWHOPE', 'EMPIRE', 'JEDI' ],
'appearsIn': ['NEWHOPE', 'EMPIRE', 'JEDI'],
'friends': [
{
'name': 'Luke Skywalker',
@ -264,11 +264,11 @@ def test_duplicate_fields():
'luke': {
'name': 'Luke Skywalker',
'homePlanet': 'Tatooine',
},
},
'leia': {
'name': 'Leia Organa',
'homePlanet': 'Alderaan',
}
}
}
result = Schema.execute(query)
assert not result.errors
@ -294,11 +294,11 @@ def test_use_fragment():
'luke': {
'name': 'Luke Skywalker',
'homePlanet': 'Tatooine',
},
},
'leia': {
'name': 'Leia Organa',
'homePlanet': 'Alderaan',
}
}
}
result = Schema.execute(query)
assert not result.errors

View File

View File

@ -0,0 +1,115 @@
from .models import Ship, Faction, Character
def initialize():
human = Character(
name='Human'
)
human.save()
droid = Character(
name='Droid'
)
droid.save()
rebels = Faction(
id='1',
name='Alliance to Restore the Republic',
hero=human
)
rebels.save()
empire = Faction(
id='2',
name='Galactic Empire',
hero=droid
)
empire.save()
xwing = Ship(
id='1',
name='X-Wing',
faction=rebels,
)
xwing.save()
ywing = Ship(
id='2',
name='Y-Wing',
faction=rebels,
)
ywing.save()
awing = Ship(
id='3',
name='A-Wing',
faction=rebels,
)
awing.save()
# Yeah, technically it's Corellian. But it flew in the service of the rebels,
# so for the purposes of this demo it's a rebel ship.
falcon = Ship(
id='4',
name='Millenium Falcon',
faction=rebels,
)
falcon.save()
homeOne = Ship(
id='5',
name='Home One',
faction=rebels,
)
homeOne.save()
tieFighter = Ship(
id='6',
name='TIE Fighter',
faction=empire,
)
tieFighter.save()
tieInterceptor = Ship(
id='7',
name='TIE Interceptor',
faction=empire,
)
tieInterceptor.save()
executor = Ship(
id='8',
name='Executor',
faction=empire,
)
executor.save()
def createShip(shipName, factionId):
nextShip = len(data['Ship'].keys())+1
newShip = Ship(
id=str(nextShip),
name=shipName
)
newShip.save()
return newShip
def getShip(_id):
return Ship.objects.get(id=_id)
def getShips():
return Ship.objects.all()
def getFaction(_id):
return Faction.objects.get(id=_id)
def getRebels():
return getFaction(1)
def getEmpire():
return getFaction(2)

View File

@ -0,0 +1,25 @@
from __future__ import absolute_import
from django.db import models
class Character(models.Model):
name = models.CharField(max_length=50)
def __str__(self):
return self.name
class Faction(models.Model):
name = models.CharField(max_length=50)
hero = models.ForeignKey(Character)
def __str__(self):
return self.name
class Ship(models.Model):
name = models.CharField(max_length=50)
faction = models.ForeignKey(Faction, related_name='ships')
def __str__(self):
return self.name

View File

@ -0,0 +1,63 @@
import graphene
from graphene import resolve_only_args, relay
from graphene.contrib.django import (
DjangoObjectType,
DjangoNode
)
from .models import (
Ship as ShipModel, Faction as FactionModel, Character as CharacterModel)
from .data import (
getFaction,
getShip,
getShips,
getRebels,
getEmpire,
)
schema = graphene.Schema(name='Starwars Django Relay Schema')
class Ship(DjangoNode):
class Meta:
model = ShipModel
@classmethod
def get_node(cls, id):
return Ship(getShip(id))
@schema.register
class CharacterModel(DjangoObjectType):
class Meta:
model = CharacterModel
class Faction(DjangoNode):
class Meta:
model = FactionModel
@classmethod
def get_node(cls, id):
return Faction(getFaction(id))
class Query(graphene.ObjectType):
rebels = graphene.Field(Faction)
empire = graphene.Field(Faction)
node = relay.NodeField()
ships = relay.ConnectionField(Ship, description='All the ships.')
@resolve_only_args
def resolve_ships(self):
return [Ship(s) for s in getShips()]
@resolve_only_args
def resolve_rebels(self):
return Faction(getRebels())
@resolve_only_args
def resolve_empire(self):
return Faction(getEmpire())
schema.query = Query

View File

@ -0,0 +1,49 @@
import pytest
from graphql.core import graphql
from .models import *
from .schema import schema
from .data import initialize
pytestmark = pytest.mark.django_db
def test_correct_fetch_first_ship_rebels():
initialize()
query = '''
query RebelsShipsQuery {
rebels {
name,
hero {
name
}
ships(first: 1) {
edges {
node {
name
}
}
}
}
}
'''
expected = {
'rebels': {
'name': 'Alliance to Restore the Republic',
'hero': {
'name': 'Human'
},
'ships': {
'edges': [
{
'node': {
'name': 'X-Wing'
}
}
]
}
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected

View File

@ -0,0 +1,119 @@
import pytest
from pytest import raises
from graphql.core import graphql
from .data import initialize
from .schema import schema
pytestmark = pytest.mark.django_db
def test_correctly_fetches_id_name_rebels():
initialize()
query = '''
query RebelsQuery {
rebels {
id
name
}
}
'''
expected = {
'rebels': {
'id': 'RmFjdGlvbjox',
'name': 'Alliance to Restore the Republic'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_refetches_rebels():
initialize()
query = '''
query RebelsRefetchQuery {
node(id: "RmFjdGlvbjox") {
id
... on Faction {
name
}
}
}
'''
expected = {
'node': {
'id': 'RmFjdGlvbjox',
'name': 'Alliance to Restore the Republic'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_fetches_id_name_empire():
initialize()
query = '''
query EmpireQuery {
empire {
id
name
}
}
'''
expected = {
'empire': {
'id': 'RmFjdGlvbjoy',
'name': 'Galactic Empire'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_refetches_empire():
initialize()
query = '''
query EmpireRefetchQuery {
node(id: "RmFjdGlvbjoy") {
id
... on Faction {
name
}
}
}
'''
expected = {
'node': {
'id': 'RmFjdGlvbjoy',
'name': 'Galactic Empire'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_refetches_xwing():
initialize()
query = '''
query XWingRefetchQuery {
node(id: "U2hpcDox") {
id
... on Ship {
name
}
}
}
'''
expected = {
'node': {
'id': 'U2hpcDox',
'name': 'X-Wing'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected

View File

View File

@ -0,0 +1,102 @@
from collections import namedtuple
Ship = namedtuple('Ship', ['id', 'name'])
Faction = namedtuple('Faction', ['id', 'name', 'ships'])
xwing = Ship(
id='1',
name='X-Wing',
)
ywing = Ship(
id='2',
name='Y-Wing',
)
awing = Ship(
id='3',
name='A-Wing',
)
# Yeah, technically it's Corellian. But it flew in the service of the rebels,
# so for the purposes of this demo it's a rebel ship.
falcon = Ship(
id='4',
name='Millenium Falcon',
)
homeOne = Ship(
id='5',
name='Home One',
)
tieFighter = Ship(
id='6',
name='TIE Fighter',
)
tieInterceptor = Ship(
id='7',
name='TIE Interceptor',
)
executor = Ship(
id='8',
name='Executor',
)
rebels = Faction(
id='1',
name='Alliance to Restore the Republic',
ships=['1', '2', '3', '4', '5']
)
empire = Faction(
id='2',
name='Galactic Empire',
ships=['6', '7', '8']
)
data = {
'Faction': {
'1': rebels,
'2': empire
},
'Ship': {
'1': xwing,
'2': ywing,
'3': awing,
'4': falcon,
'5': homeOne,
'6': tieFighter,
'7': tieInterceptor,
'8': executor
}
}
def createShip(shipName, factionId):
nextShip = len(data['Ship'].keys())+1
newShip = Ship(
id=str(nextShip),
name=shipName
)
data['Ship'][newShip.id] = newShip
data['Faction'][factionId].ships.append(newShip.id)
return newShip
def getShip(_id):
return data['Ship'][_id]
def getFaction(_id):
return data['Faction'][_id]
def getRebels():
return rebels
def getEmpire():
return empire

View File

@ -0,0 +1,54 @@
import graphene
from graphene import resolve_only_args, relay
from .data import (
getFaction,
getShip,
getRebels,
getEmpire,
)
schema = graphene.Schema(name='Starwars Relay Schema')
class Ship(relay.Node):
'''A ship in the Star Wars saga'''
name = graphene.StringField(description='The name of the ship.')
@classmethod
def get_node(cls, id):
return Ship(getShip(id))
class Faction(relay.Node):
'''A faction in the Star Wars saga'''
name = graphene.StringField(description='The name of the faction.')
ships = relay.ConnectionField(
Ship, description='The ships used by the faction.')
@resolve_only_args
def resolve_ships(self, **kwargs):
return [Ship(getShip(ship)) for ship in self.instance.ships]
@classmethod
def get_node(cls, id):
return Faction(getFaction(id))
class Query(graphene.ObjectType):
rebels = graphene.Field(Faction)
empire = graphene.Field(Faction)
node = relay.NodeField()
@resolve_only_args
def resolve_rebels(self):
return Faction(getRebels())
@resolve_only_args
def resolve_empire(self):
return Faction(getEmpire())
schema.query = Query

View File

@ -0,0 +1,61 @@
import graphene
from graphene import resolve_only_args, relay
from .data import (
getHero, getHuman, getCharacter, getDroid,
Human as _Human, Droid as _Droid)
Episode = graphene.Enum('Episode', dict(
NEWHOPE=4,
EMPIRE=5,
JEDI=6
))
def wrap_character(character):
if isinstance(character, _Human):
return Human(character)
elif isinstance(character, _Droid):
return Droid(character)
class Character(graphene.Interface):
name = graphene.StringField()
friends = relay.Connection('Character')
appearsIn = graphene.ListField(Episode)
def resolve_friends(self, args, *_):
return [wrap_character(getCharacter(f)) for f in self.instance.friends]
class Human(relay.Node, Character):
homePlanet = graphene.StringField()
class Droid(relay.Node, Character):
primaryFunction = graphene.StringField()
class Query(graphene.ObjectType):
hero = graphene.Field(Character,
episode=graphene.Argument(Episode))
human = graphene.Field(Human,
id=graphene.Argument(graphene.String))
droid = graphene.Field(Droid,
id=graphene.Argument(graphene.String))
node = relay.NodeField()
@resolve_only_args
def resolve_hero(self, episode):
return wrap_character(getHero(episode))
@resolve_only_args
def resolve_human(self, id):
return wrap_character(getHuman(id))
@resolve_only_args
def resolve_droid(self, id):
return wrap_character(getDroid(id))
Schema = graphene.Schema(query=Query)

View File

@ -0,0 +1,38 @@
from pytest import raises
from graphql.core import graphql
from .schema import schema
def test_correct_fetch_first_ship_rebels():
query = '''
query RebelsShipsQuery {
rebels {
name,
ships(first: 1) {
edges {
node {
name
}
}
}
}
}
'''
expected = {
'rebels': {
'name': 'Alliance to Restore the Republic',
'ships': {
'edges': [
{
'node': {
'name': 'X-Wing'
}
}
]
}
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected

View File

@ -0,0 +1,110 @@
from pytest import raises
from graphql.core import graphql
from .schema import schema
def test_correctly_fetches_id_name_rebels():
query = '''
query RebelsQuery {
rebels {
id
name
}
}
'''
expected = {
'rebels': {
'id': 'RmFjdGlvbjox',
'name': 'Alliance to Restore the Republic'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_refetches_rebels():
query = '''
query RebelsRefetchQuery {
node(id: "RmFjdGlvbjox") {
id
... on Faction {
name
}
}
}
'''
expected = {
'node': {
'id': 'RmFjdGlvbjox',
'name': 'Alliance to Restore the Republic'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_fetches_id_name_empire():
query = '''
query EmpireQuery {
empire {
id
name
}
}
'''
expected = {
'empire': {
'id': 'RmFjdGlvbjoy',
'name': 'Galactic Empire'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_refetches_empire():
query = '''
query EmpireRefetchQuery {
node(id: "RmFjdGlvbjoy") {
id
... on Faction {
name
}
}
}
'''
expected = {
'node': {
'id': 'RmFjdGlvbjoy',
'name': 'Galactic Empire'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected
def test_correctly_refetches_xwing():
query = '''
query XWingRefetchQuery {
node(id: "U2hpcDox") {
id
... on Ship {
name
}
}
}
'''
expected = {
'node': {
'id': 'U2hpcDox',
'name': 'X-Wing'
}
}
result = schema.execute(query)
assert not result.errors
assert result.data == expected

3
tests/utils.py Normal file
View File

@ -0,0 +1,3 @@
def assert_equal_lists(l1, l2):
assert sorted(l1) == sorted(l2)

10
tox.ini
View File

@ -1,12 +1,20 @@
[tox]
envlist = py27
envlist = py27,py33,py34,py35,pypy
[testenv]
deps=
pytest>=2.7.2
django>=1.8.0,<1.9
pytest-django
graphql-core==0.1a0
graphql-relay==0.2.0
flake8
six
blinker
singledispatch
commands=
py.test
flake8
[pytest]
DJANGO_SETTINGS_MODULE = tests.django_settings