Skip to content

Instantly share code, notes, and snippets.

@huntie
Last active October 7, 2018 10:55
Embed
What would you like to do?
Implementation of a GraphQL HTTP server for Graphene as a Falcon API resource
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment