Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
Falcon-RESTQL: POC of an on-demand field serialization using REST à la GraphQL
# Virtualenv: python3.5 -m venv ~/.virtualenvs/falcon-restql
# Activate: source ~/.virtualenvs/falcon-restql/bin/activate
# Install: pip3.5 install --upgrade cython falcon redis-limpyd httpie gunicorn
# Launch: gunicorn falcon-restql:app --reload + redis-server
# Test:
# $ http :8000/things/bicycle
# => []
# $ http POST :8000/things/bicycle name="bicycle" color="red"
# => location: /things/bicycle
# $ http :8000/things/bicycle
# => {"color": "red", "name": "bicycle"}
# $ http :8000/things/bicycle X-Fields:color
# => {"color": "red"}
import json
import falcon
from limpyd import model
main_database = model.RedisDatabase(
class ThingModel(model.RedisModel):
database = main_database
name = model.InstanceHashField(indexable=True, unique=True)
color = model.InstanceHashField()
class StorageEngine:
def get_things(self, name):
return ThingModel.collection(name=name).instances()[0]
def add_thing(self, name, color):
thing = ThingModel(name=name)
return thing
class StorageError(Exception):
def handle(ex, req, resp, params):
description = ('Sorry, couldn\'t write your thing to the '
'database. It worked on my box.')
raise falcon.HTTPError(falcon.HTTP_725,
'Database Error',
class RequireJSON:
def process_request(self, req, resp):
if not req.client_accepts_json:
raise falcon.HTTPNotAcceptable(
'This API only supports responses encoded as JSON.',
if req.method in ('POST', 'PUT'):
if 'application/json' not in req.content_type:
raise falcon.HTTPUnsupportedMediaType(
'This API only supports requests encoded as JSON.',
class JSONTranslator:
def process_request(self, req, resp):
if req.content_length in (None, 0):
# Nothing to do
body =
if not body:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
req.context['doc'] = json.loads(body.decode('utf-8'))
except (ValueError, UnicodeDecodeError):
raise falcon.HTTPError(falcon.HTTP_753,
'Malformed JSON',
'Could not decode the request body. The '
'JSON was incorrect or not encoded as '
def process_response(self, req, resp, resource):
if 'result' not in req.context:
resp.body = json.dumps(req.context['result'])
class ThingsResource:
def __init__(self, db):
self.db = db
def on_get(self, req, resp, thing_name):
result = self.db.get_things(thing_name)
except IndexError:
req.context['result'] = []
except Exception:
description = ('Aliens have attacked our base! We will '
'be back as soon as we fight them off. '
'We appreciate your patience.')
raise falcon.HTTPServiceUnavailable(
'Service Outage',
keys = set(['color', 'name'])
# This is were the filtering happens given the X-Fields header.
if req.get_header('X-Fields'):
keys &= set(req.get_header('X-Fields').split(','))
values = result.hmget(*keys)
req.context['result'] = dict(zip(keys, values))
resp.status = falcon.HTTP_200
def on_post(self, req, resp, thing_name):
doc = req.context['doc']
except KeyError:
raise falcon.HTTPBadRequest(
'Missing thing',
'A thing must be submitted in the request body.')
self.db.add_thing(doc['name'], doc['color'])
resp.status = falcon.HTTP_201
resp.location = '/things/{name}'.format(name=thing_name)
app = falcon.API(middleware=[
db = StorageEngine()
things = ThingsResource(db)
app.add_route('/things/{thing_name}', things)
app.add_error_handler(StorageError, StorageError.handle)

davidbgk commented Oct 24, 2015

Next step: pass a X-GraphQL header with a GraphQL query and filter using Graphene or a more minimalist implementation.

Nice! So you chose the X-Fields approach with flat list ?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment