Implementation of a GraphQL HTTP server for Graphene as a Falcon API resource

graphqlresource.py

import falcon
from typing import Tuple
import ujson
from .graphql import schema

class GraphQLNoQueryResolved(falcon.HTTPBadRequest):
    """
    An exception indicating that a GraphQL query could not be resolved from the
    incoming request. This will create an HTTP 400 response when raised.
    """

    def __init__(self):
        super().__init__(description='Could not resolve a GraphQL query from the request data.')

def set_base_headers(req, resp):
    resp.set_header('Allow', 'GET, POST, OPTIONS')

@falcon.after(set_base_headers)
class GraphQLResource:
    """
    Resource endpoint for accessing the application's GraphQL API.

    This implements a GraphQL HTTP server as described in
    http://graphql.org/learn/serving-over-http.
    """

    def on_options(self, req, resp):
        resp.status = falcon.HTTP_204

    def on_get(self, req, resp):
        if not 'query' in req.params:
            raise GraphQLNoQueryResolved()

        self.on_post(req, resp)

    def on_post(self, req, resp):
        query, options = self._resolve_query(req)
        result = schema.execute(query, **options)

        resp.content_type = 'application/json'

        if result.data:
            resp.body = ujson.dumps({'data': result.data})
        elif result.errors:
            resp.status = falcon.HTTP_UNPROCESSABLE_ENTITY
            resp.body = ujson.dumps({'errors': map(str, result.errors)})
        else:
            resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR

    def _resolve_query(self, req) -> Tuple[str, dict]:
        """
        Attempt to resolve the GraphQL query and any execution options from the
        request object.

        :raises GraphQLNoQueryResolved:
        """
        if req.params and 'query' in req.params:
            data = req.params
        elif req.content_type and req.content_length:
            if req.content_type == 'application/json':
                data = ujson.load(req.stream)
            elif req.content_type == 'application/graphql':
                data = {'query': req.stream.read().decode('utf-8')}
            else:
                raise falcon.HTTPUnsupportedMediaType()

        try:
            query = data['query']
        except:
            raise GraphQLNoQueryResolved()

        options = {
            'operation_name': data.get('operationName'),
            'variable_values': data.get('variables')
        }

        return (query, options)