Improved relay resolvers

This commit is contained in:
Syrus Akbary 2015-10-20 21:32:15 -07:00
parent 526d34d009
commit 464002c2db
8 changed files with 111 additions and 69 deletions

View File

@ -73,8 +73,7 @@ class Field(object):
@wraps(resolve_fn)
def custom_resolve_fn(instance, args, info):
custom_fn = getattr(instance, custom_resolve_fn_name)
return custom_fn(args, info)
return resolve_fn(instance, args, info)
return custom_resolve_fn
def get_object_type(self, schema):

View File

@ -1,5 +1,5 @@
from graphene.utils import cached_property
from collections import OrderedDict
from collections import OrderedDict, namedtuple
DEFAULT_NAMES = ('description', 'name', 'interface',
'type_name', 'interfaces', 'proxy')
@ -59,6 +59,10 @@ class Options(object):
del self.meta
@cached_property
def object(self):
return namedtuple(self.type_name, self.fields_map.keys())
def add_field(self, field):
self.local_fields.append(field)

View File

@ -113,17 +113,23 @@ class ObjectTypeMeta(type):
class BaseObjectType(object):
def __new__(cls, instance=None, *args, **kwargs):
def __new__(cls, instance=None, **kwargs):
if cls._meta.interface:
raise Exception("An interface cannot be initialized")
if instance is None:
return None
if not kwargs:
return None
elif type(instance) is cls:
instance = instance.instance
return super(BaseObjectType, cls).__new__(cls, *args, **kwargs)
def __init__(self, instance):
return super(BaseObjectType, cls).__new__(cls)
def __init__(self, instance=None, **kwargs):
signals.pre_init.send(self.__class__, instance=instance)
assert instance or kwargs
if not instance:
init_kwargs = dict({k: None for k in self._meta.fields_map.keys()}, **kwargs)
instance = self._meta.object(**init_kwargs)
self.instance = instance
signals.post_init.send(self.__class__, instance=self)

View File

@ -4,7 +4,7 @@ from graphql_relay.connection.arrayconnection import (
connection_from_list
)
from graphql_relay.connection.connection import (
connectionArgs
connection_args
)
from graphql_relay.node.node import (
from_global_id
@ -21,25 +21,45 @@ from graphene.utils import memoize
class ConnectionField(Field):
def __init__(self, field_type, resolve=None, description='', connection_type=None, edge_type=None, **kwargs):
from graphene.relay.types import Connection, Edge
def __init__(self, field_type, resolve=None, description='',
connection_type=None, edge_type=None, **kwargs):
super(ConnectionField, self).__init__(field_type, resolve=resolve,
args=connectionArgs, description=description, **kwargs)
self.connection_type = connection_type or Connection
self.edge_type = edge_type or Edge
assert issubclass(self.connection_type, Connection), 'connection_type in %r must be a subclass of Connection' % self
assert issubclass(self.edge_type, Edge), 'edge_type in %r must be a subclass of Edge' % self
args=connection_args,
description=description, **kwargs)
self.connection_type = connection_type
self.edge_type = edge_type
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)
from graphene.relay.types import PageInfo
schema = info.schema.graphene_schema
orig_resolved = super(ConnectionField, self).resolve(instance, args, info)
if orig_resolved:
resolved = self.wrap_resolved(orig_resolved, instance, args, info)
assert isinstance(
resolved, Iterable), 'Resolved value from the connection field have to be iterable'
return connection_from_list(resolved, args)
node = self.get_object_type(schema)
connection_type = self.get_connection_type(node)
edge_type = self.get_edge_type(node)
connection = connection_from_list(resolved, args, connection_type=connection_type,
edge_type=edge_type, pageinfo_type=PageInfo)
connection.set_connection_data(orig_resolved)
return connection
@memoize
def get_connection_type(self, node):
connection_type = self.connection_type or node.get_connection_type()
edge_type = self.get_edge_type(node)
return connection_type.for_node(node, edge_type=edge_type)
@memoize
def get_edge_type(self, node):
return self.edge_type or node.get_edge_type()
@memoize
def internal_type(self, schema):
@ -47,9 +67,8 @@ class ConnectionField(Field):
node = self.get_object_type(schema)
assert is_node(node), 'Only nodes have connections.'
schema.register(node)
edge_node_type = self.edge_type.for_node(node)
connection_node_type = self.connection_type.for_node(node, edge_type=edge_node_type)
return connection_node_type.internal_type(schema)
return self.get_connection_type(node).internal_type(schema)
class NodeField(Field):

View File

@ -1,9 +1,6 @@
from graphql_relay.node.node import (
to_global_id
)
from graphql_relay.connection.connection import (
connection_definitions
)
from graphene.core.types import Interface, ObjectType
from graphene.core.fields import BooleanField, StringField, ListField, Field
@ -11,15 +8,55 @@ from graphene.relay.fields import GlobalIDField
from graphene.utils import memoize
class BaseNode(object):
class PageInfo(ObjectType):
has_next_page = BooleanField(required=True, description='When paginating forwards, are there more items?')
has_previous_page = BooleanField(required=True, description='When paginating backwards, are there more items?')
start_cursor = StringField(description='When paginating backwards, the cursor to continue.')
end_cursor = StringField(description='When paginating forwards, the cursor to continue.')
class Edge(ObjectType):
'''An edge in a connection.'''
class Meta:
type_name = 'DefaultEdge'
node = Field(lambda field: field.object_type.node_type, description='The item at the end of the edge')
cursor = StringField(required=True, description='A cursor for use in pagination')
@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
def for_node(cls, node):
from graphene.relay.utils import is_node
assert is_node(node), 'ObjectTypes in a edge have to be Nodes'
return type('%s%s' % (node._meta.type_name, cls._meta.type_name), (cls, ), {'node_type': node})
class Connection(ObjectType):
'''A connection to a list of items.'''
class Meta:
type_name = 'DefaultConnection'
page_info = Field(PageInfo, required=True, description='The Information to aid in pagination')
edges = ListField(lambda field: field.object_type.edge_type, description='Information to aid in pagination.')
_connection_data = None
@classmethod
@memoize
def for_node(cls, node, edge_type=None):
from graphene.relay.utils import is_node
edge_type = edge_type or Edge
assert is_node(node), 'ObjectTypes in a connection have to be Nodes'
return type('%s%s' % (node._meta.type_name, cls._meta.type_name), (cls, ), {'edge_type': edge_type.for_node(node)})
def set_connection_data(self, data):
self._connection_data = data
def get_connection_data(self):
return self._connection_data
class BaseNode(object):
@classmethod
def _prepare_class(cls):
from graphene.relay.utils import is_node
@ -32,41 +69,18 @@ class BaseNode(object):
type_name = cls._meta.type_name
return to_global_id(type_name, instance.id)
connection_type = Connection
edge_type = Edge
@classmethod
def get_connection_type(cls):
return cls.connection_type
@classmethod
def get_edge_type(cls):
return cls.edge_type
class Node(BaseNode, Interface):
'''An object with an ID'''
id = GlobalIDField()
class PageInfo(ObjectType):
has_next_page = BooleanField(required=True, description='When paginating forwards, are there more items?')
has_previous_page = BooleanField(required=True, description='When paginating backwards, are there more items?')
start_cursor = StringField(description='When paginating backwards, the cursor to continue.')
end_cursor = StringField(description='When paginating forwards, the cursor to continue.')
class Edge(ObjectType):
'''An edge in a connection.'''
node = Field(lambda field: field.object_type.node_type, description='The item at the end of the edge')
end_cursor = StringField(required=True, description='A cursor for use in pagination')
@classmethod
@memoize
def for_node(cls, node):
from graphene.relay.utils import is_node
assert is_node(node), 'ObjectTypes in a edge have to be Nodes'
return type('%sEdge' % node._meta.type_name, (cls, ), {'node_type': node})
class Connection(ObjectType):
'''A connection to a list of items.'''
page_info = Field(PageInfo, required=True, description='The Information to aid in pagination')
edges = ListField(lambda field: field.object_type.edge_type, description='Information to aid in pagination.')
@classmethod
@memoize
def for_node(cls, node, edge_type=None):
from graphene.relay.utils import is_node
edge_type = edge_type or Edge
assert is_node(node), 'ObjectTypes in a connection have to be Nodes'
return type('%sConnection' % node._meta.type_name, (cls, ), {'edge_type': edge_type.for_node(node)})

View File

@ -57,7 +57,7 @@ setup(
'six>=1.10.0',
'blinker',
'graphql-core==0.4.7b0',
'graphql-relay==0.2.0'
'graphql-relay==0.3.3'
],
tests_require=[
'pytest>=2.7.2',

View File

@ -24,8 +24,8 @@ def test_field_no_contributed_raises_error():
def test_node_should_have_same_connection_always():
s = object()
connection1 = OtherNode.get_connection(s)
connection2 = OtherNode.get_connection(s)
connection1 = relay.Connection.for_node(OtherNode)
connection2 = relay.Connection.for_node(OtherNode)
assert connection1 == connection2

View File

@ -7,7 +7,7 @@ deps=
django>=1.8.0,<1.9
pytest-django
graphql-core==0.4.7b0
graphql-relay==0.2.0
graphql-relay==0.3.3
six
blinker
singledispatch