graphene-django/graphene_django/views.py

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'

    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()