Source code for django_graph_api.graphql.request

from graphql.ast import (
from graphql import exceptions as graphql_exceptions
from graphql.parser import GraphQLParser

from django_graph_api.graphql.utils import GraphQLError
from .validation import (

[docs]class Request(object):
[docs] def __init__(self, document, schema, variables=None, operation_name=None): """ Creates a Request object that can be validated and executed. :param document: The query string to execute. e.g. ``"query episodeNames { episodes { name } }"`` :param schema: A Schema object to run the query against :param variables: A ``dict`` of variables to pass to the query (optional) :param operation_name: If the document contains multiple named queries, the name of the query to execute (optional) """ self.document = document self.variables = variables or {} self.operation_name = operation_name self.schema = schema self._validated = False self._errors = [] self.query_root = None parser = GraphQLParser() if not self.document: self._errors.append(GraphQLError('Must provide query string.')) else: try: self.ast = parser.parse(self.document) except (graphql_exceptions.LexerError, graphql_exceptions.SyntaxError) as e: self.ast = None self._errors.append(GraphQLError( 'Parse error: {}'.format(e), line=e.line, column=e.column, )) # Additional errors are meaningless if we couldn't parse the document if self._errors: self._validated = True
@property def errors(self): if self._validated: return self._errors self.validate() return self._errors @property def operation(self): if not hasattr(self, '_operation'): self._set_operation_and_fragments() if self._operation is None: raise Exception('Request document has no valid operation') return self._operation @property def fragments(self): if not hasattr(self, '_fragments'): self._set_fragments_and_fragments() return self._fragments def _set_operation_and_fragments(self): # This is a bit duplicative, but allows accessing these properties # without running validation. if self.ast is None: raise Exception('Request document is unparseable') operations = {} self._fragments = {} for definition in self.ast.definitions: if isinstance(definition, OperationDefinition): operations[] = definition elif isinstance(definition, FragmentDefinition): self._fragments[] = definition try: if self.operation_name: self._operation = operations[self.operation_name] else: self._operation = list(operations.values())[0] except (KeyError, IndexError): self._operation = None
[docs] def validate(self): """ Used to perform validation of a query before execution. Errors produced from validation can be accessed from ``request.errors``. If a Request object has been validated once, additional calls will not re-run validation. """ if self._validated: return self._validated = True try: self._errors.extend(perform_operation_validation(self)) if self._errors: # If the request is invalid, stop validation return self.query_root = self.schema.get_query_root(self) self._errors.extend(perform_argument_validation(self.query_root)) except Exception as e: if not isinstance(e, GraphQLError): e = GraphQLError(e) self.errors.append(e)
[docs] def execute(self): """ :return: data, errors """ query_root = self.schema.get_query_root(self) data, errors = query_root.execute() self._errors.extend(errors) return data, errors