-
Notifications
You must be signed in to change notification settings - Fork 819
/
connection.py
119 lines (91 loc) · 4.03 KB
/
connection.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
import re
from collections import Iterable
import six
from graphql_relay import connection_definitions, connection_from_list
from graphql_relay.connection.connection import connection_args
from ..types.field import Field
from ..types.objecttype import ObjectType, ObjectTypeMeta
from ..types.options import Options
from ..utils.copy_fields import copy_fields
from ..utils.get_fields import get_fields
from ..utils.is_base_type import is_base_type
from ..utils.props import props
class ConnectionMeta(ObjectTypeMeta):
def __new__(cls, name, bases, attrs):
super_new = type.__new__
# Also ensure initialization is only performed for subclasses of Model
# (excluding Model class itself).
if not is_base_type(bases, ConnectionMeta):
return super_new(cls, name, bases, attrs)
options = Options(
attrs.pop('Meta', None),
name=None,
description=None,
node=None,
interfaces=[],
)
Edge = attrs.pop('Edge', None)
edge_fields = props(Edge) if Edge else {}
edge_fields = get_fields(ObjectType, edge_fields, ())
connection_fields = copy_fields(Field, get_fields(ObjectType, attrs, bases))
cls = super_new(cls, name, bases, dict(attrs, _meta=options))
assert options.node, 'You have to provide a node in {}.Meta'.format(cls.__name__)
from ..utils.get_graphql_type import get_graphql_type
edge, connection = connection_definitions(
name=options.name or re.sub('Connection$', '', cls.__name__),
node_type=get_graphql_type(options.node),
resolve_node=cls.resolve_node,
resolve_cursor=cls.resolve_cursor,
edge_fields=edge_fields,
connection_fields=connection_fields,
)
cls.Edge = type(edge.name, (ObjectType, ), {'Meta': type('Meta', (object,), {'graphql_type': edge})})
cls._meta.graphql_type = connection
fields = copy_fields(Field, options.graphql_type.get_fields(), parent=cls)
cls._meta.get_fields = lambda: fields
return cls
class Connection(six.with_metaclass(ConnectionMeta, ObjectType)):
resolve_node = None
resolve_cursor = None
def __init__(self, *args, **kwargs):
kwargs['pageInfo'] = kwargs.pop('pageInfo', kwargs.pop('page_info'))
super(Connection, self).__init__(*args, **kwargs)
class IterableConnectionField(Field):
def __init__(self, type, args={}, *other_args, **kwargs):
super(IterableConnectionField, self).__init__(type, args=connection_args, *other_args, **kwargs)
@property
def type(self):
from ..utils.get_graphql_type import get_graphql_type
return get_graphql_type(self.connection)
@type.setter
def type(self, value):
self._type = value
@property
def connection(self):
from .node import Node
graphql_type = super(IterableConnectionField, self).type
if issubclass(graphql_type.graphene_type, Node):
connection_type = graphql_type.graphene_type.get_default_connection()
else:
connection_type = graphql_type.graphene_type
assert issubclass(connection_type, Connection), '{} type have to be a subclass of Connection'.format(str(self))
return connection_type
def connection_resolver(self, root, args, context, info):
iterable = super(ConnectionField, self).resolver(root, args, context, info)
# if isinstance(resolved, self.type.graphene)
assert isinstance(
iterable, Iterable), 'Resolved value from the connection field have to be iterable. Received "{}"'.format(iterable)
connection = connection_from_list(
iterable,
args,
connection_type=self.connection,
edge_type=self.connection.Edge,
)
return connection
@property
def resolver(self):
return self.connection_resolver
@resolver.setter
def resolver(self, resolver):
self._resolver = resolver
ConnectionField = IterableConnectionField