import inspect import json import re import six from django.http import HttpResponse, HttpResponseNotAllowed from django.http.response import HttpResponseBadRequest from django.shortcuts import render from django.views.generic import View from graphql import Source, execute, parse, validate from graphql.error import format_error as format_graphql_error from graphql.error import GraphQLError from graphql.execution import ExecutionResult from graphql.execution.middleware import MiddlewareManager from graphql.type.schema import GraphQLSchema from graphql.utils.get_operation_ast import get_operation_ast from .settings import graphene_settings class HttpError(Exception): def __init__(self, response, message=None, *args, **kwargs): self.response = response self.message = message = message or response.content.decode() super(HttpError, self).__init__(message, *args, **kwargs) def get_accepted_content_types(request): def qualify(x): parts = x.split(';', 1) if len(parts) == 2: match = re.match(r'(^|;)q=(0(\.\d{,3})?|1(\.0{,3})?)(;|$)', parts[1]) if match: return parts[0], float(match.group(2)) return parts[0], 1 raw_content_types = request.META.get('HTTP_ACCEPT', '*/*').split(',') qualified_content_types = map(qualify, raw_content_types) return list(x[0] for x in sorted(qualified_content_types, key=lambda x: x[1], reverse=True)) def instantiate_middleware(middlewares): for middleware in middlewares: if inspect.isclass(middleware): yield middleware() continue yield middleware class GraphQLView(View): graphiql_version = '0.7.8' graphiql_template = 'graphene/graphiql.html' schema = None graphiql = False executor = None middleware = None root_value = None pretty = False def __init__(self, schema=None, executor=None, middleware=None, root_value=None, graphiql=False, pretty=False): if not schema: schema = graphene_settings.SCHEMA if middleware is None: middleware = graphene_settings.MIDDLEWARE self.schema = schema if middleware is not None: self.middleware = MiddlewareManager(*list(instantiate_middleware(middleware))) self.executor = executor self.root_value = root_value self.pretty = pretty self.graphiql = graphiql assert isinstance(self.schema, GraphQLSchema), 'A Schema is required to be provided to GraphQLView.' # noinspection PyUnusedLocal def get_root_value(self, request): return self.root_value def get_middleware(self, request): return self.middleware def get_context(self, request): return request def dispatch(self, request, *args, **kwargs): try: if request.method.lower() not in ('get', 'post'): raise HttpError(HttpResponseNotAllowed(['GET', 'POST'], 'GraphQL only supports GET and POST requests.')) data = self.parse_body(request) show_graphiql = self.graphiql and self.can_display_graphiql(request, data) query, variables, operation_name = self.get_graphql_params(request, data) execution_result = self.execute_graphql_request( request, data, query, variables, operation_name, show_graphiql ) if execution_result: response = {} if execution_result.errors: response['errors'] = [self.format_error(e) for e in execution_result.errors] if execution_result.invalid: status_code = 400 else: status_code = 200 response['data'] = execution_result.data result = self.json_encode(request, response, pretty=show_graphiql) else: result = None if show_graphiql: return self.render_graphiql( request, graphiql_version=self.graphiql_version, query=query or '', variables=variables or '', operation_name=operation_name or '', result=result or '' ) return HttpResponse( status=status_code, content=result, content_type='application/json' ) except HttpError as e: response = e.response response['Content-Type'] = 'application/json' response.content = self.json_encode(request, { 'errors': [self.format_error(e)] }) return response def render_graphiql(self, request, **data): return render(request, self.graphiql_template, data) def json_encode(self, request, d, pretty=False): if not (self.pretty or pretty) and not request.GET.get('pretty'): return json.dumps(d, separators=(',', ':')) return json.dumps(d, sort_keys=True, indent=2, separators=(',', ': ')) # noinspection PyBroadException def parse_body(self, request): content_type = self.get_content_type(request) if content_type == 'application/graphql': return {'query': request.body.decode()} elif content_type == 'application/json': try: request_json = json.loads(request.body.decode('utf-8')) assert isinstance(request_json, dict) return request_json except: raise HttpError(HttpResponseBadRequest('POST body sent invalid JSON.')) elif content_type in ['application/x-www-form-urlencoded', 'multipart/form-data']: return request.POST return {} def execute(self, *args, **kwargs): return execute(self.schema, *args, **kwargs) def execute_graphql_request(self, request, data, query, variables, operation_name, show_graphiql=False): if not query: if show_graphiql: return None raise HttpError(HttpResponseBadRequest('Must provide query string.')) source = Source(query, name='GraphQL request') try: document_ast = parse(source) validation_errors = validate(self.schema, document_ast) if validation_errors: return ExecutionResult( errors=validation_errors, invalid=True, ) except Exception as e: return ExecutionResult(errors=[e], invalid=True) if request.method.lower() == 'get': operation_ast = get_operation_ast(document_ast, operation_name) if operation_ast and operation_ast.operation != 'query': if show_graphiql: return None raise HttpError(HttpResponseNotAllowed( ['POST'], 'Can only perform a {} operation from a POST request.'.format(operation_ast.operation) )) try: return self.execute( document_ast, root_value=self.get_root_value(request), variable_values=variables, operation_name=operation_name, context_value=self.get_context(request), middlewares=self.get_middleware(request), executor=self.executor, ) except Exception as e: return ExecutionResult(errors=[e], invalid=True) @classmethod def can_display_graphiql(cls, request, data): raw = 'raw' in request.GET or 'raw' in data return not raw and cls.request_wants_html(request) @classmethod def request_wants_html(cls, request): accepted = get_accepted_content_types(request) html_index = accepted.count('text/html') json_index = accepted.count('application/json') return html_index > json_index @staticmethod def get_graphql_params(request, data): query = request.GET.get('query') or data.get('query') variables = request.GET.get('variables') or data.get('variables') if variables and isinstance(variables, six.text_type): try: variables = json.loads(variables) except: raise HttpError(HttpResponseBadRequest('Variables are invalid JSON.')) operation_name = request.GET.get('operationName') or data.get('operationName') return query, variables, operation_name @staticmethod def format_error(error): if isinstance(error, GraphQLError): return format_graphql_error(error) return {'message': six.text_type(error)} @staticmethod def get_content_type(request): meta = request.META content_type = meta.get('CONTENT_TYPE', meta.get('HTTP_CONTENT_TYPE', '')) return content_type.split(';', 1)[0].lower()