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
|
import inspect
from graphql import GraphQLObjectType, GraphQLSchema, graphql, is_type
from graphql.type.directives import (
GraphQLDirective,
GraphQLIncludeDirective,
GraphQLSkipDirective,
)
from graphql.type.introspection import IntrospectionSchema
from graphql.utils.introspection_query import introspection_query
from graphql.utils.schema_printer import print_schema
from .definitions import GrapheneGraphQLType
from .objecttype import ObjectType
from .typemap import TypeMap, is_graphene_type
def assert_valid_root_type(_type):
if _type is None:
return
is_graphene_objecttype = inspect.isclass(_type) and issubclass(_type, ObjectType)
is_graphql_objecttype = isinstance(_type, GraphQLObjectType)
assert is_graphene_objecttype or is_graphql_objecttype, (
"Type {} is not a valid ObjectType."
).format(_type)
class Schema(GraphQLSchema):
"""
Graphene Schema can execute operations (query, mutation, subscription) against the defined
types.
For advanced purposes, the schema can be used to lookup type definitions and answer questions
about the types through introspection.
Args:
query (ObjectType): Root query *ObjectType*. Describes entry point for fields to *read*
data in your Schema.
mutation (ObjectType, optional): Root mutation *ObjectType*. Describes entry point for
fields to *create, update or delete* data in your API.
subscription (ObjectType, optional): Root subscription *ObjectType*. Describes entry point
for fields to receive continuous updates.
directives (List[GraphQLDirective], optional): List of custom directives to include in
GraphQL schema. Defaults to only include directives definved by GraphQL spec (@include
and @skip) [GraphQLIncludeDirective, GraphQLSkipDirective].
types (List[GraphQLType], optional): List of any types to include in schema that
may not be introspected through root types.
auto_camelcase (bool): Fieldnames will be transformed in Schema's TypeMap from snake_case
to camelCase (preferred by GraphQL standard). Default True.
"""
def __init__(
self,
query=None,
mutation=None,
subscription=None,
directives=None,
types=None,
auto_camelcase=True,
):
assert_valid_root_type(query)
assert_valid_root_type(mutation)
assert_valid_root_type(subscription)
self._query = query
self._mutation = mutation
self._subscription = subscription
self.types = types
self.auto_camelcase = auto_camelcase
if directives is None:
directives = [GraphQLIncludeDirective, GraphQLSkipDirective]
assert all(
isinstance(d, GraphQLDirective) for d in directives
), "Schema directives must be List[GraphQLDirective] if provided but got: {}.".format(
directives
)
self._directives = directives
self.build_typemap()
def get_query_type(self):
return self.get_graphql_type(self._query)
def get_mutation_type(self):
return self.get_graphql_type(self._mutation)
def get_subscription_type(self):
return self.get_graphql_type(self._subscription)
def __getattr__(self, type_name):
"""
This function let the developer select a type in a given schema
by accessing its attrs.
Example: using schema.Query for accessing the "Query" type in the Schema
"""
_type = super(Schema, self).get_type(type_name)
if _type is None:
raise AttributeError('Type "{}" not found in the Schema'.format(type_name))
if isinstance(_type, GrapheneGraphQLType):
return _type.graphene_type
return _type
def get_graphql_type(self, _type):
if not _type:
return _type
if is_type(_type):
return _type
if is_graphene_type(_type):
graphql_type = self.get_type(_type._meta.name)
assert graphql_type, "Type {} not found in this schema.".format(
_type._meta.name
)
assert graphql_type.graphene_type == _type
return graphql_type
raise Exception("{} is not a valid GraphQL type.".format(_type))
def execute(self, *args, **kwargs):
"""
Use the `graphql` function from `graphql-core` to provide the result for a query string.
Most of the time this method will be called by one of the Graphene :ref:`Integrations`
via a web request.
Args:
request_string (str or Document): GraphQL request (query, mutation or subscription) in
string or parsed AST form from `graphql-core`.
root (Any, optional): Value to use as the parent value object when resolving root
types.
context (Any, optional): Value to be made avaiable to all resolvers via
`info.context`. Can be used to share authorization, dataloaders or other
information needed to resolve an operation.
variables (dict, optional): If variables are used in the request string, they can be
provided in dictionary form mapping the variable name to the variable value.
operation_name (str, optional): If mutiple operations are provided in the
request_string, an operation name must be provided for the result to be provided.
middleware (List[SupportsGraphQLMiddleware]): Supply request level middleware as
defined in `graphql-core`.
backend (GraphQLCoreBackend, optional): Override the default GraphQLCoreBackend.
**execute_options (Any): Depends on backend selected. Default backend has several
options such as: validate, allow_subscriptions, return_promise, executor.
Returns:
:obj:`ExecutionResult` containing any data and errors for the operation.
"""
return graphql(self, *args, **kwargs)
def introspect(self):
instrospection = self.execute(introspection_query)
if instrospection.errors:
raise instrospection.errors[0]
return instrospection.data
def __str__(self):
return print_schema(self)
def lazy(self, _type):
return lambda: self.get_type(_type)
def build_typemap(self):
initial_types = [
self._query,
self._mutation,
self._subscription,
IntrospectionSchema,
]
if self.types:
initial_types += self.types
self._type_map = TypeMap(
initial_types, auto_camelcase=self.auto_camelcase, schema=self
)
|
