-
Notifications
You must be signed in to change notification settings - Fork 822
/
types.py
176 lines (137 loc) · 5.65 KB
/
types.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
import inspect
import warnings
from collections import Iterable
from functools import wraps
import six
from graphql_relay.connection.arrayconnection import connection_from_list
from graphql_relay.node.node import to_global_id
from ..core.classtypes import InputObjectType, Interface, Mutation, ObjectType
from ..core.classtypes.interface import InterfaceMeta
from ..core.classtypes.mutation import MutationMeta
from ..core.types import Boolean, Field, List, String
from ..core.types.argument import ArgumentsGroup
from ..core.types.definitions import NonNull
from ..utils import memoize
from .fields import GlobalIDField
class PageInfo(ObjectType):
has_next_page = Boolean(
required=True,
description='When paginating forwards, are there more items?')
has_previous_page = Boolean(
required=True,
description='When paginating backwards, are there more items?')
start_cursor = String(
description='When paginating backwards, the cursor to continue.')
end_cursor = String(
description='When paginating forwards, the cursor to continue.')
class Edge(ObjectType):
'''An edge in a connection.'''
cursor = String(
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'
node_field = Field(node, description='The item at the end of the edge')
return type(
'%s%s' % (node._meta.type_name, cls._meta.type_name),
(cls,),
{'node_type': node, 'node': node_field})
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')
_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.for_node(node)
assert is_node(node), 'ObjectTypes in a connection have to be Nodes'
edges = List(edge_type, description='Information to aid in pagination.')
return type(
'%s%s' % (node._meta.type_name, cls._meta.type_name),
(cls,),
{'edge_type': edge_type, 'edges': edges})
@classmethod
def from_list(cls, iterable, args, info):
assert isinstance(
iterable, Iterable), 'Resolved value from the connection field have to be iterable'
connection = connection_from_list(
iterable, args, connection_type=cls,
edge_type=cls.edge_type, pageinfo_type=PageInfo)
connection.set_connection_data(iterable)
return connection
def set_connection_data(self, data):
self._connection_data = data
def get_connection_data(self):
return self._connection_data
class NodeMeta(InterfaceMeta):
def construct_get_node(cls):
get_node = getattr(cls, 'get_node', None)
assert get_node, 'get_node classmethod not found in %s Node' % cls
assert callable(get_node), 'get_node have to be callable'
args = 3
if isinstance(get_node, staticmethod):
args -= 1
get_node_num_args = len(inspect.getargspec(get_node).args)
if get_node_num_args < args:
warnings.warn("get_node will receive also the info arg"
" in future versions of graphene".format(cls.__name__),
FutureWarning)
@staticmethod
@wraps(get_node)
def wrapped_node(*node_args):
if len(node_args) < args:
node_args += (None, )
return get_node(*node_args[:-1])
setattr(cls, 'get_node', wrapped_node)
def construct(cls, *args, **kwargs):
cls = super(NodeMeta, cls).construct(*args, **kwargs)
if not cls._meta.abstract:
cls.construct_get_node()
return cls
class Node(six.with_metaclass(NodeMeta, Interface)):
'''An object with an ID'''
id = GlobalIDField()
class Meta:
abstract = True
def to_global_id(self):
type_name = self._meta.type_name
return to_global_id(type_name, self.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 MutationInputType(InputObjectType):
client_mutation_id = String(required=True)
class RelayMutationMeta(MutationMeta):
def construct(cls, *args, **kwargs):
cls = super(RelayMutationMeta, cls).construct(*args, **kwargs)
if not cls._meta.abstract:
assert hasattr(
cls, 'mutate_and_get_payload'), 'You have to implement mutate_and_get_payload'
return cls
def construct_arguments(cls, items):
new_input_type = type('{}Input'.format(
cls._meta.type_name), (MutationInputType, ), items)
cls.add_to_class('input_type', new_input_type)
return ArgumentsGroup(input=NonNull(new_input_type))
class ClientIDMutation(six.with_metaclass(RelayMutationMeta, Mutation)):
client_mutation_id = String(required=True)
class Meta:
abstract = True
@classmethod
def mutate(cls, instance, args, info):
input = args.get('input')
payload = cls.mutate_and_get_payload(input, info)
client_mutation_id = input.get('client_mutation_id')
setattr(payload, 'client_mutation_id', client_mutation_id)
return payload